You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
112 KiB
112 KiB
AgentScope 企业级多租户 AI 平台 - 完整技术方案
版本: v2.0 (已升级为 AgentScope Runtime 架构)
适用规模: 200 人公司
部署方式: Docker Compose 全容器化
核心原则: 不修改 AgentScope 和 AgentScope Runtime 源码，通过扩展和组合实现所有功能
关键决策: 网关层采用 AgentScope Runtime 的 AgentApp（而非裸 FastAPI），节省 7000+ 行基础设施代码
目录
- 前置分析：AgentScope + Runtime 现状
- 核心决策：AgentScope Runtime vs 裸 FastAPI
- 整体架构总览
- 前端项目规划
- 后端架构设计（基于 AgentApp 扩展，不改源码）
- 双 RBAC 权限体系详细设计
- 四大业务场景落地方案
- Dify-like 可视化流编排引擎
- 数据库设计
- Docker 部署方案
- API 接口设计
- 开发路线图
- 项目目录结构
一、前置分析：AgentScope + Runtime 现状
1.1 AgentScope 有没有前端？
没有。 AgentScope 是一个纯 Python 后端框架/库，通过 pip install agentscope 安装使用。不包含任何前端 UI 代码。
但有一个配套的外部项目 AgentScope Studio（独立部署，非 agentscope 包的一部分），用于运行时监控和追踪。它通过 HTTP 钩子（hooks/_studio_hooks.py）将消息推送到 Studio 展示，但 Studio 本身：
- 是额外安装的项目，非 agentscope 核心
- 仅用于开发者调试/监控，不面向最终用户
- 不具备业务管理、用户管理、工作流编排等企业功能
1.2 AgentScope 的核心能力清单
| 能力 | 对应的类/模块 | 我们如何使用 |
|---|---|---|
| ReAct 智能体 | ReActAgent |
直接继承，创建业务 Agent |
| 智能体基类 | AgentBase |
直接继承，创建流节点 Agent |
| 工具管理 | Toolkit |
直接使用，注册业务工具 |
| 流水线编排 | SequentialPipeline, FanoutPipeline, MsgHub |
代码级可用，需上层封装为动态引擎 |
| MCP 协议 | MCPClientBase, StdIOStatefulClient, HttpStatefulClient |
直接使用，连接外部服务 |
| 消息系统 | Msg, TextBlock, ImageBlock, 等 |
直接使用 |
| 记忆系统 | InMemoryMemory, Mem0LongTermMemory, 等 |
包装使用，添加用户隔离层 |
| RAG | KnowledgeBase, SimpleKnowledge |
直接使用 |
| 钩子系统 | pre/post reply/print/observe/reasoning/acting | 注册自定义钩子，注入 RBAC 上下文 |
| 格式化器 | OpenAIChatFormatter, 等 |
直接使用 |
| 模型接入 | OpenAIChatModel, DashScopeChatModel, 等 |
直接使用 |
| 结构化输出 | structured_model 参数 |
直接使用 |
| 追踪 | OpenTelemetry | 直接使用 |
1.3 是否可以完全不修改 AgentScope 源码？
可以。 核心论证：
┌─────────────────────────────────────────────────────────┐
│ 所有业务代码写在一个新项目中 │
│ │
│ our-enterprise-platform/ │
│ ├── pyproject.toml # 依赖 agentscope │
│ ├── backend/ # 我们的后端代码 │
│ │ ├── agents/ # 继承 AgentScope 的类 │
│ │ ├── flow_engine/ # 流编排引擎 │
│ │ └── ... │
│ └── frontend/ # 我们的前端代码 │
│ │
│ pip install agentscope ← 作为普通依赖安装，不修改源码 │
└─────────────────────────────────────────────────────────┘
AgentScope 的每个模块都有清晰的基类和扩展点，我们只需要：
| 扩展方式 | 原理 | 是否需要改 AgentScope |
|---|---|---|
继承 ReActAgent / AgentBase |
Python 标准继承 | ❌ 不需要 |
使用 Toolkit().register_tool_function() |
注册工具函数 | ❌ 不需要 |
使用 register_class_hook() / register_instance_hook() |
注册钩子 | ❌ 不需要 |
继承 FormatterBase |
扩展格式化器 | ❌ 不需要 |
使用 MCPClientBase 子类 |
MCP 协议接入 | ❌ 不需要 |
包装 MemoryBase |
记忆层封装 | ❌ 不需要 |
直接使用 KnowledgeBase |
RAG 能力 | ❌ 不需要 |
唯一需要额外开发的是：AgentScope 的 Pipeline 是 Python 代码级的（必须在代码中写 SequentialPipeline([agent1, agent2])），要实现 Dify 那样的可视化拖拽编排，需要我们在上层建立一个流引擎（Flow Engine），读取 JSON 流定义，动态组装 AgentScope 组件。这是"封装"而非"修改"。
1.4 AgentScope Runtime 是什么？
AgentScope Runtime 是 AgentScope 的官方生产级网关运行时，它是一个独立项目（本地代码），负责将 AgentScope 智能体部署为标准的 HTTP 服务。
核心架构：
agentscope-runtime/
├── src/agentscope_runtime/
│ ├── engine/
│ │ ├── app/
│ │ │ └── agent_app.py # ★ AgentApp 类 - 继承 FastAPI
│ │ ├── runner.py # Runner - 智能体执行引擎
│ │ ├── deployers/ # 多种部署器
│ │ │ ├── adapter/a2a/ # A2A 协议适配器
│ │ │ ├── adapter/agui/ # AG-UI (Web聊天界面)
│ │ │ ├── adapter/responses/ # OpenAI兼容 ResponseAPI
│ │ │ └── utils/service_utils/
│ │ │ └── routing/
│ │ │ ├── unified_routing_mixin.py # 统一路由管理
│ │ │ ├── custom_endpoint_mixin.py # 自定义端点支持
│ │ │ └── task_engine_mixin.py # Celery任务引擎
│ │ ├── schemas/ # AgentRequest/AgentResponse 等
│ │ ├── tracing/ # OpenTelemetry 追踪
│ │ └── adapters/ # 多框架适配器
│ │ ├── agentscope/ # AgentScope 框架适配
│ │ ├── langgraph/ # LangGraph 框架适配
│ │ ├── agno/ # Agno 框架适配
│ │ └── autogen/ # AutoGen 框架适配
│ └── sandbox/ # 沙箱执行环境
│ ├── filesystem/ # 文件系统沙箱
│ ├── browser/ # 浏览器沙箱
│ └── gui/ # GUI 沙箱
1.5 AgentApp 的关键设计
# agent_app.py 第 60 行 - 核心发现
class AgentApp(FastAPI, UnifiedRoutingMixin, InterruptMixin):
"""Agent application integrating FastAPI and Runner"""
关键事实：AgentApp 直接继承自 FastAPI！ 这意味着：
-
它是一个完整的 FastAPI 实例 - 可以使用所有 FastAPI 能力：
@app.get(),@app.post()等装饰器添加路由app.add_middleware()添加中间件- FastAPI 依赖注入 (
Depends) app.include_router()包含子路由- WebSocket 支持
-
额外获得的 Runtime 能力：
UnifiedRoutingMixin- 统一路由管理、自定义端点、Celery 任务InterruptMixin- 中断/取消正在执行的 Agent 任务- 内置
/health,/process,/admin/*端点 - A2A、AG-UI、ResponseAPI 三协议自动适配
- 会话管理 (session_id)
- OTEL 分布式追踪
-
已有的扩展接口：
lifespan参数 - 注入自定义启动/关闭逻辑custom_endpoints参数 - 注册自定义路由（支持热恢复）before_start/after_finish回调protocol_adapters- 自定义协议适配器
二、核心决策：AgentScope Runtime vs 裸 FastAPI
2.1 深度对比分析
经过对 AgentScope Runtime 源码的完整分析（agent_app.py、runner.py、routing 模块），给出以下对比：
| 能力维度 | AgentScope Runtime (AgentApp) | 裸 FastAPI 自建 |
|---|---|---|
| FastAPI 所有能力 | ✅ 完全可用（AgentApp IS FastAPI） | ✅ 完全可用 |
| 智能体执行引擎 | ✅ Runner 内置，支持流式/非流式 | ❌ 需自己实现（~500行） |
| SSE 流式输出 | ✅ 内置 StreamingResponse + _stream_generator |
❌ 需自己实现（~300行） |
| 会话管理 | ✅ AgentRequest.session_id + user_id 内置 |
❌ 需自己实现（~200行） |
| A2A 协议 | ✅ 内置 A2AFastAPIDefaultAdapter | ❌ 需自己实现（~1000行） |
| AG-UI Web聊天 | ✅ 内置 AGUIDefaultAdapter，可直接跑 Web UI | ❌ 需自己实现 |
| OpenAI兼容API | ✅ 内置 ResponseAPIDefaultAdapter | ❌ 需自己实现（~500行） |
| 中断/取消 | ✅ InterruptMixin + Redis/Local backend | ❌ 需自己实现（~400行） |
| Celery 后台任务 | ✅ TaskEngineMixin 内置 | ❌ 需自己实现（~300行） |
| OTEL 追踪 | ✅ 内置 tracing 模块 | ❌ 需自己集成（~200行） |
| 多框架适配 | ✅ AgentScope/LangGraph/AutoGen/Agno | ❌ 需自己实现 |
| 沙箱执行 | ✅ Browser/FS/GUI/Mobile 沙箱 | ❌ 无 |
| K8s/Knative 部署 | ✅ 多 Deployer 内置 | ❌ 需自己实现 |
| 自定义端点 | ✅ endpoint() / task() 装饰器 + custom_endpoints |
✅ @app.get/post() |
| 自定义中间件 | ✅ app.add_middleware() |
✅ app.add_middleware() |
| 自定义路由 | ✅ @app.get() 直接可用 |
✅ @app.get() 直接可用 |
| 鉴权/Auth | ❌ 无（user_id 只是透传） |
❌ 无，需自己加 |
| RBAC | ❌ 无 | ❌ 无，需自己加 |
| DB ORM | ❌ 无 | ❌ 无，需自己加 |
2.2 节省工作量估算
使用 AgentScope Runtime 而非裸 FastAPI，节省约 7000+ 行生产级基础设施代码：
| 组件 | 自建代码量 | Runtime 提供 | 是否可以不改 Runtime |
|---|---|---|---|
| Agent SSE 流式服务 | ~500 行 | ✅ | 不需要（直接可用） |
| 会话/用户管理 | ~300 行 | ✅ | 不需要（AgentRequest 内置） |
| A2A 协议适配 | ~1000 行 | ✅ | 不需要（A2A Adapter 内置） |
| OpenAI 兼容 API | ~500 行 | ✅ | 不需要（Response API Adapter） |
| 中断/取消 | ~400 行 | ✅ | 不需要（InterruptMixin） |
| 后台任务 | ~300 行 | ✅ | 不需要（Celery TaskEngine） |
| 追踪/监控 | ~200 行 | ✅ | 不需要（OTEL tracing） |
| 健康检查/管理 | ~300 行 | ✅ | 不需要（/health, /admin/*） |
| 多框架适配 | ~2000 行 | ✅ | 不需要（多 Adapter 内置） |
| 总计 | ~5500 行 | ✅ 全有 | 全部零修改可用 |
2.3 最终结论
┌──────────────────────────────────────────────────────────────────┐
│ │
│ ★★★ 结 论: 使用 AgentScope Runtime (AgentApp) ★★★ │
│ │
│ 原因: │
│ 1. AgentApp IS FastAPI - 拥有 FastAPI 所有能力 │
│ 2. 零修改可用的基础设施 - 7000+ 行业务代码节省 │
│ 3. 不改 Runtime 源码 - 通过 FastAPI 原生装饰器扩展 │
│ 4. 无任何技术负担 - 可以随时绕过 Runtime 直接用 FastAPI 底层 │
│ 5. 获得未来升级能力 - Runtime 升级自动获得新协议/框架支持 │
│ │
│ 我们的代码是 AgentScope Runtime 的外挂模块， │
│ 类似 Express.js 中间件的思路：核心框架不动，业务逻辑外挂。 │
│ │
└──────────────────────────────────────────────────────────────────┘
2.4 扩展方式验证（不改 Runtime 源码）
# 这是我们项目的入口文件 - 完全不修改 agentscope_runtime 源码
from contextlib import asynccontextmanager
from agentscope_runtime import AgentApp
# ─── 自定义启动/关闭（通过 lifespan）───
@asynccontextmanager
async def enterprise_lifespan(app):
# 启动: 初始化数据库连接池、加载RBAC缓存、注册企微回调
await init_db_pool()
await load_rbac_cache()
await init_wecom_client()
yield
# 关闭: 清理资源
await close_db_pool()
# ─── 创建 AgentApp（不改 Runtime）───
app = AgentApp(
app_name="Enterprise AI Platform",
app_description="企业级多租户AI平台",
lifespan=enterprise_lifespan, # ← 注入自定义生命周期
mode="daemon_thread",
)
# ─── 添加自定义 FastAPI 路由（AgentApp IS FastAPI，原生语法即可）───
@app.get("/api/org/departments")
async def get_departments(user: dict = Depends(get_current_user)):
"""组织架构 - 部门树"""
return await org_service.get_departments()
@app.post("/api/auth/login")
async def login(credentials: LoginRequest):
"""用户登录"""
return await auth_service.login(credentials)
@app.get("/api/monitor/employee/{emp_id}/analysis")
async def get_employee_analysis(emp_id: str, user: dict = Depends(require_manager)):
"""AI分析员工工作"""
return await monitor_service.analyze(emp_id, manager_id=user["id"])
# ─── 添加自定义中间件 ───
@app.middleware("http")
async def rbac_middleware(request: Request, call_next):
"""RBAC 权限拦截中间件"""
token = request.headers.get("Authorization", "").replace("Bearer ", "")
user_context = await jwt_decode(token)
request.state.user = user_context
return await call_next(request)
# ─── 使用 Runtime 的内置 custom_endpoints 机制 ───
@app.endpoint("/api/wecom/callback", methods=["POST"])
async def wecom_callback(request: dict):
"""企业微信回调接收"""
return await wecom_service.handle_callback(request)
@app.endpoint("/api/flows/{flow_id}/execute", methods=["POST"])
async def execute_flow(request: dict, flow_id: str):
"""执行流编排"""
return await flow_engine.execute(flow_id, request)
# ─── 启动 ───
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8000, web_ui=True) # web_ui=True 启用AG-UI聊天界面
验证结论：所有扩展均通过 FastAPI 原生机制（装饰器/中间件/依赖注入/lifespan）实现，完全不修改 agentscope_runtime 一行源码。
三、整体架构总览
3.1 系统分层图（Runtime 版）
┌─────────────────────────────────────────────────────────────────────┐
│ 用户触达层 │
│ ┌──────────────────────┐ ┌───────────────────────────┐ │
│ │ 企业微信机器人 │ │ 管理后台 Web │ │
│ │ (员工日常交流/AI助手) │ │ (领导看板/工作流编排) │ │
│ └──────────┬───────────┘ └─────────────┬─────────────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌──────────────────────────────────────────────────────────────┐ │
│ │ Nginx 网关层 │ │
│ │ · 路由分发 (管理端 /api/* | 企微回调 /wecom/*) │ │
│ │ · HTTPS | 限流 | 静态文件服务 │ │
│ └──────────────────────────┬───────────────────────────────────┘ │
│ ▼ │
├─────────────────────────────────────────────────────────────────────┤
│ AgentScope Runtime 内核 (AgentApp IS FastAPI) │
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ 内置协议适配器 │ │ Runner 引擎 │ │ 会话/中断管理 │ │
│ │ A2A/AG-UI/ │ │ 流式/非流式 │ │ session_id │ │
│ │ ResponseAPI │ │ 多框架支持 │ │ 中断/恢复 │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │
│ ┌────────────────── 我们的业务扩展 (FastAPI 原生注册) ──────────┐ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌────────────────┐ │ │
│ │ │ Auth/JWT │ │ 双 RBAC │ │ 企微网关 │ │ 流编排引擎 │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └────────────────┘ │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌────────────────┐ │ │
│ │ │ 任务管理 │ │ 效能监控 │ │ MCP注册 │ │ 审计日志 │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └────────────────┘ │ │
│ └──────────────────────────────────────────────────────────────┘ │
│ │ │
│ ┌──────────────────────────────────────────────────────────────┐ │
│ │ AgentScope 集成层 (继承封装，不改源码) │ │
│ │ 业务Agent | RBAC Hook | 记忆隔离 | 自定义工具 │ │
│ └──────────────────────────────────────────────────────────────┘ │
│ │ │
├─────────────────────────────┼───────────────────────────────────────┤
│ 数据与存储层 │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────────┐ │
│ │PostgreSQL│ │ Redis │ │ Qdrant/ │ │ MinIO │ │
│ │ 业务数据 │ │ 缓存/会话 │ │ Milvus │ │ 文件/文档存储 │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘
3.2 关键设计决策
| 决策点 | 方案 | 理由 |
|---|---|---|
| AgentScope 源码 | 零修改 | pip install, 通过继承、钩子、封装实现所有功能 |
| AgentScope Runtime | 零修改 | AgentApp 继承 FastAPI，通过 @app.get/post 等原生机制扩展 |
| 网关层 | AgentScope Runtime AgentApp | 节省 7000+ 行基础设施代码，内置 SSE/会话/A2A/AG-UI |
| 框架关系 | agentscope + agentscope-runtime 作为 pip 依赖 | 类似 Spring 依赖三方库，核心框架可独立升级 |
| 前端 | Vue 3 管理端 + 企微 Bot（无独立前端） | 管理端 SPA 独立部署，员工直接在企微中使用 |
| 前后端通信 | RESTful API + SSE + WebSocket | 管理端用 REST，Agent 推理用 SSE，实时消息用 WS |
| 服务间通信 | Docker 网络 + 内部 HTTP | 容器间标准通信 |
| 身份认证 | JWT + 企业微信 OAuth | 统一认证，通过 FastAPI Depends 注入 |
| 部署 | Docker Compose 全容器化 | 一键部署，环境一致 |
四、前端项目规划
4.1 双入口架构：为什么不是一个前端项目？
4.1.1 核心结论：双入口 · 单前端代码
双 RBAC ≠ 两套前端代码
双 RBAC = 两个入口 + 同一套前端代码 + 两层权限过滤
| 入口 | 形态 | 使用角色 | 交互方式 | RBAC 体现 |
|---|---|---|---|---|
| 入口A: 企微 Bot | 企业微信内置 | 全员（员工为主） | AI对话/文件/卡片 | 组织RBAC：Agent自动隔离数据 |
| 入口B: Web 管理后台 | Vue 3 SPA | 全员（不同角色不同菜单） | 表单/图表/拖拽编辑器 | 平台RBAC控制菜单+组织RBAC控制数据 |
4.1.2 为什么不做两个独立前端项目？
❌ 错误理解： ✅ 正确架构：
员工前端项目 ────→ 员工API 同一套 Vue 代码
│ │ │
└─完全独立─┐ ┌─完全独立─┘ router.beforeEach
管理前端项目 ────→ 管理API ├── 员工 → 只看到个人中心
├── 经理 → 多出监控/任务/分析
这种会导致： ├── HR → 多出组织架构管理
· 登录页 ×2 ├── 编辑 → 多出流编排
· 用户信息 ×2 └── 超管 → 全部可见
· 组织树组件 ×2
· 维护成本翻倍 同一套登录、用户信息、组织树组件
4.1.3 双 RBAC 在两个入口中的具体体现
| 入口A: 企微 Bot | 入口B: Web 管理后台 | |
|---|---|---|
| 平台 RBAC (功能权限) | 控制 Agent 可注册的工具集合 | 控制菜单/按钮/路由可见性 |
| 员工Agent → 文档处理+知识查询工具 | 员工登录 → 仅显示「个人中心」 | |
| 经理Agent → 额外获得下属分析工具 | 经理登录 → 多出「监控/任务/分析」 | |
| 组织 RBAC (数据权限) | 记忆系统天然按 user_id 隔离 | API 层 SQL 自动附加 department 过滤 |
| 问"我的任务"→ 只查自己的 task 记录 | 经理看下属列表 → 仅显示本部门下属 | |
| 实现方式 | AgentFactory 创建时注入 user_context | Vue Router 守卫 + API 中间件 |
4.1.4 Web 管理后台：同一套代码如何实现角色分化
// Vue Router - 平台RBAC通过路由元信息控制
const routes = [
{ path: '/dashboard', meta: { perms: ['*'] }, component: Dashboard },
{ path: '/my-center', meta: { perms: ['self:read'] }, component: MyCenter }, // 人人可见
{ path: '/monitor', meta: { perms: ['monitor:read'] }, component: Monitor }, // 仅领导
{ path: '/org/users', meta: { perms: ['user:read'] }, component: UserList }, // 仅HR/超管
{ path: '/flow/editor', meta: { perms: ['flow:create'] }, component: FlowEditor }, // 仅编辑者
{ path: '/audit', meta: { perms: ['audit:read'] }, component: AuditLog }, // 仅超管
]
router.beforeEach((to, from, next) => {
const userPerms = store.state.user.permissions
if (to.meta.perms && !hasAny(userPerms, to.meta.perms)) {
next('/403')
}
next()
})
// 侧边栏 - 同一组件，按角色动态渲染
<Sidebar>
<MenuItem v-if="can('self:read')" to="/my-center"> 个人中心</MenuItem>
<MenuItem v-if="can('monitor:read')" to="/monitor"> 工作监控</MenuItem>
<MenuItem v-if="can('flow:create')" to="/flow/editor">流编排</MenuItem>
<MenuItem v-if="can('user:read')" to="/org/users"> 人员管理</MenuItem>
<MenuItem v-if="can('audit:read')" to="/audit"> 审计日志</MenuItem>
</Sidebar>
4.1.5 端到端隔离验证：跨部门经理无法看到非下属数据
部门经理A（市场部）打开 /monitor/employees
│
├── 平台RBAC: can('monitor:read')? → ✅ 通过
│
├── 前端侧边栏: 显示"工作监控"菜单
│
├── GET /api/monitor/subordinates
│ → 后端依赖注入: manager_id = user_A
│ → 组织RBAC: get_all_subordinates(user_A)
│ → 返回: [市场部员工1, 市场部员工2] ← 天然不含研发部员工
│
└── 页面渲染: 只能看到市场部的下属
部门经理B（研发部）打开同一个页面
│
├── 平台RBAC: can('monitor:read')? → ✅ 通过（同样的菜单）
│
├── GET /api/monitor/subordinates
│ → 组织RBAC: get_all_subordinates(user_B)
│ → 返回: [研发部员工3, 研发部员工4]
│
└── 看到的是完全不同的数据 ← 组织RBAC发挥作用
小结：同一套前端代码 + 同一个 URL + 同一个 API → 不同人看到完全不同的人和数据。这就是双 RBAC 的精髓——不是靠物理隔离项目，而是靠逻辑隔离权限。
4.2 管理端前端详细设计
技术栈: Vue 3 + TypeScript + Vite + Element Plus + Pinia + Vue Router
admin-frontend/
├── src/
│ ├── views/ # 页面视图
│ │ ├── login/ # 登录页
│ │ ├── dashboard/ # 首页概览
│ │ ├── org/ # 组织架构管理
│ │ │ ├── DepartmentTree.vue # 部门树
│ │ │ ├── UserList.vue # 用户列表
│ │ │ └── UserDetail.vue # 用户详情/编辑
│ │ ├── role/ # 角色权限管理
│ │ │ ├── RoleList.vue # 角色列表
│ │ │ ├── PermissionConfig.vue # 权限配置
│ │ │ └── DataScopeConfig.vue # 数据范围配置
│ │ ├── monitor/ # 员工工作监控
│ │ │ ├── EmployeeList.vue # 下属列表
│ │ │ ├── WorkDashboard.vue # 工作看板
│ │ │ └── AIAnalysis.vue # AI分析报告
│ │ ├── task/ # 任务管理
│ │ │ ├── TaskCreate.vue # 创建任务
│ │ │ ├── TaskList.vue # 任务列表
│ │ │ └── TaskDetail.vue # 任务详情
│ │ ├── flow/ # 流编排
│ │ │ ├── FlowEditor.vue # 可视化流编辑器 (核心)
│ │ │ ├── FlowList.vue # 流列表
│ │ │ └── FlowMarket.vue # 流上架市场
│ │ ├── wecom/ # 企业微信配置
│ │ │ ├── BotConfig.vue # 机器人配置
│ │ │ └── MessageTemplate.vue # 消息模板
│ │ ├── agent/ # 智能体管理
│ │ │ ├── AgentList.vue # 智能体列表
│ │ │ └── AgentChat.vue # 智能体对话测试
│ │ └── audit/ # 审计日志
│ │ └── AuditLog.vue
│ ├── components/ # 公共组件
│ │ ├── layout/ # 布局组件
│ │ └── common/ # 通用组件
│ ├── stores/ # Pinia 状态管理
│ ├── api/ # API 请求封装
│ └── router/ # 路由配置
└── package.json
4.3 管理端路由结构
/login → 登录页
/dashboard → 首页仪表盘
/org → 组织架构
/org/departments → 部门管理
/org/users → 人员管理
/org/users/:id → 人员详情
/role → 角色权限
/role/list → 角色列表
/role/:id/permissions → 权限配置
/monitor → 工作监控
/monitor/employees → 下属列表
/monitor/:id/dashboard → 个人工作看板
/monitor/:id/analysis → AI分析报告
/task → 任务管理
/task/create → 创建任务
/task/list → 任务列表
/flow → 流编排
/flow/editor → 流编辑器（核心）
/flow/editor/:id → 编辑已有流
/flow/list → 流列表
/flow/market → 已上架流
/wecom → 企微配置
/wecom/bot → 机器人配置
/agent → 智能体
/agent/list → 智能体列表
/agent/chat/:id → 对话测试
/audit → 审计日志
4.4 企微员工端的交互形态
企业微信 App 内:
┌──────────────────────────┐
│ 公司 AI 助手 │
│ │
│ 🧑 员工小王 │
│ 早上好！有什么可以帮你？ │
│ │
│ 员工: 帮我看看上周的项目 │
│ 进度报告格式对不对 │
│ │
│ 员工: [上传文件] │
│ 项目进度报告.docx │
│ │
│ AI: 好的，正在处理... │
│ │
│ AI: 已完成格式修正: │
│ · 标题层级已统一 │
│ · 表格格式已规范化 │
│ · 已导入业务系统 │
│ [下载修正后文件] │
│ │
│ ────── 新任务通知 ────── │
│ 📋 领导张三 给你安排了 │
│ 新任务: │
│ "完成Q2市场调研报告" │
│ 截止: 2026-05-20 │
│ [查看详情] [确认收到] │
└──────────────────────────┘
五、后端架构设计（基于 AgentApp 扩展，不改源码）
5.1 后端服务整体架构
backend/
├── main.py # FastAPI 入口
├── config.py # 全局配置
├── dependencies.py # 依赖注入
│
├── modules/ # 业务模块
│ ├── auth/ # 认证鉴权
│ │ ├── router.py # 登录/注册/刷新Token API
│ │ ├── service.py # 认证业务逻辑
│ │ ├── models.py # 用户数据模型
│ │ └── jwt_handler.py # JWT 签发/验证
│ │
│ ├── rbac/ # 双RBAC引擎
│ │ ├── router.py # 角色/权限 CRUD API
│ │ ├── service.py # 权限校验逻辑
│ │ ├── models.py # 角色/权限/部门模型
│ │ ├── platform_rbac.py # 平台RBAC (功能权限)
│ │ └── org_rbac.py # 组织RBAC (数据权限)
│ │
│ ├── wecom/ # 企业微信网关
│ │ ├── router.py # 企微回调接收 API
│ │ ├── service.py # 消息路由/处理
│ │ ├── crypto.py # 企微消息加解密
│ │ └── models.py # 企微数据模型
│ │
│ ├── agent_manager/ # 智能体管理
│ │ ├── router.py # 智能体 CRUD API
│ │ ├── service.py # 智能体生命周期管理
│ │ └── pool.py # 智能体实例池
│ │
│ ├── flow_engine/ # 流编排引擎 (Dify-like)
│ │ ├── router.py # 流 CRUD + 执行 API
│ │ ├── service.py # 流生命周期管理
│ │ ├── engine.py # 核心: JSON → AgentScope Pipeline
│ │ ├── node_factory.py # 节点工厂(创建各类节点Agent)
│ │ ├── validator.py # 流定义校验
│ │ └── models.py # 流定义数据模型
│ │
│ ├── task/ # 任务管理
│ │ ├── router.py # 任务 CRUD API
│ │ ├── service.py # 任务生命周期
│ │ └── models.py # 任务数据模型
│ │
│ ├── monitor/ # 工作监控与分析
│ │ ├── router.py # 看板/分析 API
│ │ ├── service.py # 分析业务逻辑
│ │ └── models.py # 分析结果模型
│ │
│ ├── mcp_registry/ # MCP 服务注册中心
│ │ ├── router.py # MCP 注册/发现 API
│ │ ├── service.py # MCP 服务管理
│ │ └── models.py # MCP 连接配置模型
│ │
│ └── audit/ # 审计日志
│ ├── router.py # 审计查询 API
│ └── service.py # 日志记录
│
├── agentscope_integration/ # AgentScope 集成层 (核心)
│ ├── agents/ # 自定义 Agent 类
│ │ ├── employee_agent.py # 员工 AI 助手 Agent
│ │ ├── manager_agent.py # 管理分析 Agent
│ │ ├── task_agent.py # 任务管理 Agent
│ │ ├── flow_node_agent.py # 流节点通用 Agent
│ │ └── document_agent.py # 文档处理 Agent
│ │
│ ├── hooks/ # 钩子注册
│ │ ├── rbac_hook.py # RBAC 上下文注入钩子
│ │ ├── audit_hook.py # 审计日志钩子
│ │ └── context_hook.py # 用户上下文注入
│ │
│ ├── memory/ # 记忆隔离封装
│ │ └── user_memory.py # 按用户隔离的记忆代理
│ │
│ ├── tools/ # 自定义工具
│ │ ├── wecom_tools.py # 企微推送/通知工具
│ │ ├── document_tools.py # 文档格式修正工具
│ │ ├── org_tools.py # 组织架构查询工具
│ │ └── business_tools.py # 业务系统对接工具
│ │
│ └── factory.py # Agent 工厂 (统一创建入口)
│
├── models/ # SQLAlchemy ORM 模型
│ ├── base.py # 基类
│ ├── user.py # 用户
│ ├── department.py # 部门
│ ├── role.py # 角色
│ ├── permission.py # 权限
│ ├── task.py # 任务
│ ├── flow_definition.py # 流定义
│ └── audit_log.py # 审计日志
│
├── schemas/ # Pydantic 请求/响应模型
│ └── ...
│
├── middleware/ # FastAPI 中间件
│ ├── auth_middleware.py # 认证中间件
│ ├── rbac_middleware.py # RBAC 拦截
│ └── audit_middleware.py # 审计记录
│
└── migration/ # 数据库迁移 (Alembic)
└── versions/
5.2 AgentScope 集成层核心设计
这层是整个系统最关键的部分，决定了"零修改"原则的可行性。
5.2.1 Agent 工厂模式
# backend/agentscope_integration/factory.py
"""Agent 工厂 - 统一创建所有业务 Agent，所有实现基于继承而非修改"""
from agentscope.agent import ReActAgent, AgentBase
from agentscope.model import OpenAIChatModel, DashScopeChatModel
from agentscope.formatter import OpenAIChatFormatter, DashScopeChatFormatter
from agentscope.tool import Toolkit
from agentscope.memory import InMemoryMemory
# ---- 以下是我们的自定义类，全部继承自 AgentScope ----
from .agents.employee_agent import EmployeeAssistantAgent
from .agents.manager_agent import ManagerAnalysisAgent
from .agents.task_agent import TaskManagementAgent
from .agents.flow_node_agent import FlowNodeAgent
from .agents.document_agent import DocumentProcessingAgent
from .memory.user_memory import UserIsolatedMemory
from .hooks.rbac_hook import register_rbac_hooks
class AgentFactory:
"""Agent 工厂类，负责创建和缓存所有业务 Agent 实例"""
# 模型配置池（支持多模型切换）
_model_pool: dict[str, tuple] = {}
_agent_cache: dict[str, AgentBase] = {}
@classmethod
async def create_employee_agent(
cls,
user_id: str,
user_name: str,
department_id: str,
) -> EmployeeAssistantAgent:
"""
创建员工 AI 助手实例（每个员工独立）
基于 ReActAgent 继承，不修改源码
"""
model, formatter = cls._get_model("employee")
agent = EmployeeAssistantAgent(
name=f"EmployeeAI_{user_name}",
sys_prompt=cls._build_employee_sys_prompt(user_name, department_id),
model=model,
formatter=formatter,
toolkit=cls._build_employee_toolkit(user_id),
memory=UserIsolatedMemory(user_id=user_id), # 隔离的记忆
max_iters=10,
)
return agent
@classmethod
async def create_manager_agent(cls, manager_id: str) -> ManagerAnalysisAgent:
"""创建管理分析 Agent（每个领导独立）"""
# ... 类似，但 sys_prompt 注入下属列表和可访问的数据范围
pass
@classmethod
def _build_employee_sys_prompt(cls, name: str, dept_id: str) -> str:
return f"""你是 {name} 的专属 AI 工作助手。你可以:
1. 回答工作中的问题
2. 帮助处理文档、修正格式
3. 查询公司知识库获取信息
重要约束:
- 你只能访问该员工权限范围内的数据和工具
- 你的回答不会被其他员工看到
- 涉及敏感操作需要二次确认"""
@classmethod
def _build_employee_toolkit(cls, user_id: str) -> Toolkit:
"""为员工 Agent 注册工具"""
from .tools.wecom_tools import send_wecom_notification
from .tools.document_tools import process_document_format
from .tools.business_tools import query_business_system
toolkit = Toolkit()
toolkit.register_tool_function(send_wecom_notification)
toolkit.register_tool_function(process_document_format)
toolkit.register_tool_function(query_business_system)
return toolkit
5.2.2 记忆隔离封装
# backend/agentscope_integration/memory/user_memory.py
"""按用户隔离的记忆代理 - 封装而非修改 AgentScope 源码"""
from agentscope.memory import MemoryBase, InMemoryMemory
from agentscope.message import Msg
class UserIsolatedMemory(MemoryBase):
"""
在 AgentScope 原生 MemoryBase 之上加一层用户隔离。
不修改 agentscope 源码，纯粹通过继承和代理实现。
"""
def __init__(self, user_id: str, backend_memory: MemoryBase | None = None):
self.user_id = user_id
# 委托给原生 InMemoryMemory（生产环境可换 Redis/SQL）
self._backend = backend_memory or InMemoryMemory()
async def add(self, msg: Msg | list[Msg] | None) -> None:
"""写入时自动标记用户ID"""
if msg is None:
return
if isinstance(msg, list):
for m in msg:
m.metadata = m.metadata or {}
m.metadata["_user_id"] = self.user_id
else:
msg.metadata = msg.metadata or {}
msg.metadata["_user_id"] = self.user_id
await self._backend.add(msg)
async def get_memory(self, **kwargs) -> list[Msg]:
"""读取时自动过滤，只返回当前用户的数据"""
all_msgs = await self._backend.get_memory(**kwargs)
return [m for m in all_msgs if m.metadata.get("_user_id") == self.user_id]
async def delete_by_mark(self, mark: str) -> None:
await self._backend.delete_by_mark(mark)
async def update_messages_mark(self, msg_ids: list[str], new_mark: str) -> None:
await self._backend.update_messages_mark(msg_ids, new_mark)
async def update_compressed_summary(self, summary: str) -> None:
await self._backend.update_compressed_summary(summary)
5.2.3 RBAC 钩子注入
# backend/agentscope_integration/hooks/rbac_hook.py
"""RBAC 钩子 - 在 Agent 执行前后注入权限上下文"""
from agentscope.agent import AgentBase
from agentscope.message import Msg
def create_rbac_pre_reply_hook(user_context: dict):
"""
创建一个注入特定用户 RBAC 上下文的 pre_reply 钩子。
每个用户一个专属钩子，确保数据隔离。
"""
async def rbac_pre_reply_hook(self: AgentBase, kwargs: dict) -> dict:
"""在 Agent 回复前注入权限上下文到系统提示和消息中"""
msg = kwargs.get("msg")
# 注入用户身份到 metadata
if msg and isinstance(msg, Msg):
msg.metadata = msg.metadata or {}
msg.metadata["_user_id"] = user_context["user_id"]
msg.metadata["_role"] = user_context["role"]
msg.metadata["_department_id"] = user_context["department_id"]
msg.metadata["_org_path"] = user_context["org_path"]
# 动态修改系统提示，添加权限约束
if hasattr(self, "_sys_prompt"):
self._sys_prompt = self._sys_prompt + f"""
\n[权限约束]
- 当前操作者: {user_context['user_name']} ({user_context['role']})
- 部门: {user_context['department_name']}
- 可操作数据范围: {user_context['data_scope']}
- 禁止访问非授权范围内的任何数据
"""
return kwargs
return rbac_pre_reply_hook
def register_rbac_hooks(user_context: dict):
"""
为全局 AgentBase 注册 RBAC 钩子。
所有业务 Agent 自动获得权限控制能力。
"""
hook = create_rbac_pre_reply_hook(user_context)
hook_name = f"rbac_{user_context['user_id']}"
AgentBase.register_class_hook("pre_reply", hook_name, hook)
六、双 RBAC 权限体系详细设计
6.1 双 RBAC 概念模型
┌─────────────────────────────────────────────────────────────┐
│ 双 RBAC 权限体系 │
│ │
│ ┌───────────────────────┐ ┌──────────────────────────┐ │
│ │ L1: 平台 RBAC │ │ L2: 组织 RBAC │ │
│ │ (功能权限) │ │ (数据权限) │ │
│ │ │ │ │ │
│ │ 控制: 能做什么 │ │ 控制: 能看谁的数据 │ │
│ │ 粒度: 菜单/按钮/API │ │ 粒度: 行级数据过滤 │ │
│ │ │ │ │ │
│ │ 例: │ │ 例: │ │
│ │ · 能否打开"流编辑器" │ │ · 张经理能看到李四的 │ │
│ │ · 能否"上架工作流" │ │ 工作分析报告吗？ │ │
│ │ · 能否"管理用户" │ │ → 李四是张的下属吗？ │ │
│ │ │ │ │ │
│ │ 判断: role → perm │ │ 判断: user → org_path │ │
│ │ → API access │ │ → data_scope │ │
│ └───────────────────────┘ └──────────────────────────┘ │
│ │
│ 两者叠加: 张经理的角色决定了他在管理后台能看到的菜单， │
│ 他的组织位置决定了每个菜单里能看到哪些人的数据。 │
└─────────────────────────────────────────────────────────────┘
6.2 数据库模型
-- ============================================================
-- 核心表定义
-- ============================================================
-- 部门表（树形结构）
CREATE TABLE departments (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
name VARCHAR(100) NOT NULL,
parent_id UUID REFERENCES departments(id), -- 上级部门
path VARCHAR(500) NOT NULL, -- 路径: /公司/技术部/AI组
level INT NOT NULL DEFAULT 0, -- 层级
sort_order INT DEFAULT 0,
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
);
-- 用户表
CREATE TABLE users (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
username VARCHAR(50) UNIQUE NOT NULL,
password_hash VARCHAR(255) NOT NULL,
display_name VARCHAR(100) NOT NULL,
email VARCHAR(100),
phone VARCHAR(20),
wecom_user_id VARCHAR(100) UNIQUE, -- 企业微信UserID
department_id UUID REFERENCES departments(id), -- 所属部门
position VARCHAR(100), -- 岗位
manager_id UUID REFERENCES users(id), -- 直属上级
status VARCHAR(20) DEFAULT 'active', -- active/inactive
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
);
-- 角色表
CREATE TABLE roles (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
name VARCHAR(50) UNIQUE NOT NULL, -- super_admin, dept_manager, employee
description VARCHAR(200),
is_system BOOLEAN DEFAULT FALSE, -- 系统预置角色不可删除
created_at TIMESTAMP DEFAULT NOW()
);
-- 权限表（功能权限/API权限）
CREATE TABLE permissions (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
code VARCHAR(100) UNIQUE NOT NULL, -- flow:create, user:delete, report:view
name VARCHAR(100) NOT NULL,
resource VARCHAR(100) NOT NULL, -- 资源类型: flow, user, task, report
action VARCHAR(50) NOT NULL, -- 操作: create, read, update, delete, publish
description VARCHAR(200)
);
-- 角色-权限关联
CREATE TABLE role_permissions (
role_id UUID REFERENCES roles(id) ON DELETE CASCADE,
permission_id UUID REFERENCES permissions(id) ON DELETE CASCADE,
PRIMARY KEY (role_id, permission_id)
);
-- 用户-角色关联
CREATE TABLE user_roles (
user_id UUID REFERENCES users(id) ON DELETE CASCADE,
role_id UUID REFERENCES roles(id) ON DELETE CASCADE,
PRIMARY KEY (user_id, role_id)
);
-- 数据权限范围定义（组织RBAC的核心）
CREATE TABLE data_scopes (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
code VARCHAR(50) UNIQUE NOT NULL, -- self_only, subordinate, dept_only, all
name VARCHAR(100) NOT NULL,
description VARCHAR(200)
);
-- 角色-数据范围关联
CREATE TABLE role_data_scopes (
role_id UUID REFERENCES roles(id) ON DELETE CASCADE,
data_scope_id UUID REFERENCES data_scopes(id) ON DELETE CASCADE,
PRIMARY KEY (role_id, data_scope_id)
);
6.3 预置角色与权限矩阵
预置角色:
super_admin: # 超级管理员
platform_permissions: # L1: 所有功能权限
- "*:*" # 通配符，全部
data_scope: all # L2: 可看全部数据
assignable_to: [] # 仅系统初始化，不分配给普通用户
hr_admin: # HR 管理员
platform_permissions:
- user:create
- user:read
- user:update
- department:read
- department:create
- role:read
data_scope: all # 可看全员组织数据（IT人员）
assignable_to: [特定IT人员]
org_admin: # 组织管理员（如CEO/VP）
platform_permissions:
- monitor:* # 所有监控相关
- report:read
- task:create
- task:read
- flow:read
data_scope: all # 可看全公司
assignable_to: [CEO, VP]
dept_manager: # 部门经理
platform_permissions:
- monitor:read # 可查看监控页面
- employee:read # 可查看员工列表
- analysis:read # 可查看分析报告
- task:create # 可创建任务
- task:read # 可查看任务
- report:read
data_scope: subordinate_only # L2: 仅看本人+下属
assignable_to: [各部门经理]
team_lead: # 组长
platform_permissions:
- monitor:read
- employee:read
- task:create
- task:read
data_scope: subordinate_only # L2: 仅看本人+直接下属
assignable_to: [各组长]
employee: # 普通员工
platform_permissions:
- self:read # 只能看自己的信息
- task:read # 可以看分配给自己的任务
data_scope: self_only # L2: 仅看本人
assignable_to: [全体员工]
workflow_editor: # 工作流编辑者（附加角色）
platform_permissions:
- flow:create
- flow:update
- flow:read
- flow:publish # 上架到企微
data_scope: self_only
assignable_to: [IT人员/高级管理员]
预置数据范围:
self_only: 仅当前用户自己的数据
subordinate_only: 当前用户 + 直接下属 + 间接下属（递归）
dept_only: 当前部门所有数据
all: 全部数据
6.4 RBAC 权限校验流程
请求到达:
GET /api/monitor/employee/{target_user_id}/analysis
Header: Authorization: Bearer <JWT>
│
▼
┌──────────────────┐
│ 1. JWT 解析 │ → 得到 current_user_id, current_role
└────────┬─────────┘
▼
┌──────────────────┐
│ 2. 平台RBAC校验 │ → 检查 current_role 是否有 monitor:read 权限
│ (功能权限) │ → 没有? 返回 403 Forbidden
└────────┬─────────┘
▼
┌──────────────────┐
│ 3. 组织RBAC校验 │ → 检查 target_user_id 是否在 current_user 的下属链中
│ (数据权限) │ → 不在? 返回 403 "无权查看该员工数据"
└────────┬─────────┘
▼
┌──────────────────┐
│ 4. 执行请求 │ → 通过 FastAPI 依赖注入传递 user_context
│ │ → Agent 执行时自动注入权限范围
└──────────────────┘
6.5 绝对隔离的实现保证
# backend/modules/rbac/org_rbac.py
async def get_visible_user_ids(current_user_id: str) -> list[str]:
"""
组织RBAC核心: 计算当前用户可见的用户ID列表
这是"绝对隔离"的最终实现
跨部门领导看不到非下属 → 因为下属链只包含本部门人员
"""
user = await get_user(current_user_id)
role = await get_primary_role(user)
if role.data_scope_code == "all":
# CEO/HR → 返回全部用户
return await get_all_user_ids()
elif role.data_scope_code == "self_only":
# 普通员工 → 只看自己
return [current_user_id]
elif role.data_scope_code == "subordinate_only":
# 经理/组长 → 递归获取下属链
subordinates = await get_all_subordinates_recursive(current_user_id)
return [current_user_id] + [s.id for s in subordinates]
elif role.data_scope_code == "dept_only":
# 部门范围
dept_users = await get_department_users(user.department_id)
return [u.id for u in dept_users]
return []
七、四大业务场景落地方案
6.1 场景1：员工 × 企微 × AI 助手
流程详解：
员工在企微发消息 "帮我处理这个文档"
│
▼
┌──────────────────────────────────────────────────────────┐
│ Step 1: 企微回调 → API 网关 │
│ POST /api/wecom/callback │
│ Body: { user_id, msg_type, content, file_url } │
└─────────────────────┬────────────────────────────────────┘
▼
┌──────────────────────────────────────────────────────────┐
│ Step 2: 认证 + 路由 │
│ · 根据 wecom_user_id 查找系统用户 │
│ · 提取用户上下文 (user_id, role, dept) │
│ · 路由到对应的 Agent │
└─────────────────────┬────────────────────────────────────┘
▼
┌──────────────────────────────────────────────────────────┐
│ Step 3: AgentScope 员工 Agent 执行 │
│ │
│ EmployeeAssistantAgent.process(msg): │
│ │
│ ├── [推理] LLM 理解意图: "用户想修正文档格式并导入系统" │
│ │ │
│ ├── [动作1] 调用 process_document_format(file_url) │
│ │ → MCP → 文档处理服务 → 返回修正后文件 │
│ │ │
│ ├── [动作2] 调用 query_business_system( │
│ │ action="import", file=corrected_file) │
│ │ → MCP → 业务系统 │
│ │ │
│ └── [输出] "文档已修正格式并导入系统，修改项: ..." │
└─────────────────────┬────────────────────────────────────┘
▼
┌──────────────────────────────────────────────────────────┐
│ Step 4: 结果返回企微 │
│ · AgentScope Msg → 企微消息格式转换 │
│ · 通过企微 API 推送回复 │
│ · 交互记录存入记忆 (用于后续分析) │
└──────────────────────────────────────────────────────────┘
对应的 AgentScope 代码结构：
# backend/agentscope_integration/agents/employee_agent.py
from agentscope.agent import ReActAgent # 继承，不修改源码
class EmployeeAssistantAgent(ReActAgent):
"""
员工 AI 助手。
完全继承 ReActAgent，仅定制 sys_prompt 和工具集。
"""
async def reply(self, msg, structured_model=None):
"""重写 reply 以添加用户上下文注入"""
# 注入用户上下文到消息 metadata
if isinstance(msg, Msg):
msg.metadata["_user_id"] = self.user_id
msg.metadata["_wecom_msg_id"] = self.msg_id
return await super().reply(msg, structured_model)
6.2 场景2：中层管理 × 员工效能看板
流程详解：
领导在管理后台点击 "查看小王的工作分析"
│
▼
┌──────────────────────────────────────────────────────────┐
│ Step 1: 权限校验 │
│ · 平台RBAC: 领导角色有 monitor:read 权限? → Yes │
│ · 组织RBAC: 小王是领导下属? → Yes │
│ · 通过 → 进入 Step 2 │
└─────────────────────┬────────────────────────────────────┘
▼
┌──────────────────────────────────────────────────────────┐
│ Step 2: 拉取小王的AI交互记录 │
│ · 从 UserIsolatedMemory(user_id=小王) 获取对话历史 │
│ · 统计: 交互次数、活跃天数、话题分布 │
└─────────────────────┬────────────────────────────────────┘
▼
┌──────────────────────────────────────────────────────────┐
│ Step 3: LLM 分析 │
│ │
│ ManagerAnalysisAgent.analyze(employee_id=小王): │
│ │
│ 输入: 小王的全部AI交互记录 (过去7天) │
│ │
│ System Prompt: │
│ "你是一个企业管理者分析助手。请根据员工的AI交互记录， │
│ 分析以下维度并生成结构化报告:" │
│ │
│ 输出 (结构化, 使用 AgentScope 的 structured_model): │
│ { │
│ "task_completion_rate": 0.85, │
│ "active_days": 5, │
│ "main_topics": ["市场调研", "文档处理", "数据分析"], │
│ "efficiency_trend": "提升", │
│ "strengths": ["文档处理能力强", "主动使用AI工具"], │
│ "growth_suggestions": ["可学习进阶数据分析工具"], │
│ "personality_traits": "细致认真，工作效率稳定提升" │
│ } │
└─────────────────────┬────────────────────────────────────┘
▼
┌──────────────────────────────────────────────────────────┐
│ Step 4: 结果展示 │
│ · 后端返回 JSON │
│ · 前端渲染为可视化看板 (图表 + AI分析文字) │
└──────────────────────────────────────────────────────────┘
关键设计：分析隔离：
# 伪代码 - 分析流程中的数据隔离保障
async def analyze_employee(manager_id, target_employee_id):
"""分析指定员工的工作情况"""
# 1. 组织RBAC: 确认 target 是 manager 的下属
subordinates = await get_all_subordinates(manager_id)
if target_employee_id not in [s.id for s in subordinates]:
raise PermissionError("无权查看该员工数据")
# 2. 获取目标员工的交互记录（天然隔离，不会读到其他人数据）
target_memory = UserIsolatedMemory(user_id=target_employee_id)
interactions = await target_memory.get_memory()
# 3. 交给 ManagerAnalysisAgent 分析
# Agent 的 system_prompt 中已经注入了权限范围，不会越界
agent = await AgentFactory.create_manager_agent(manager_id)
analysis = await agent.analyze(interactions)
# 4. 记录审计日志
await audit_log(
operator=manager_id,
action="view_analysis",
target=target_employee_id
)
return analysis
6.3 场景3：领导 × 任务分派 × 企微推送
流程详解：
领导在管理后台 → 创建任务 → "小王，本周五前完成市场调研报告"
│
▼
┌──────────────────────────────────────────────────────────┐
│ Step 1: 权限校验 │
│ · 平台RBAC: 领导有 task:create 权限? → Yes │
│ · 组织RBAC: 小王是领导下属? → Yes │
└─────────────────────┬────────────────────────────────────┘
▼
┌──────────────────────────────────────────────────────────┐
│ Step 2: 写入任务表 │
│ INSERT INTO tasks ( │
│ assigner_id, assignee_id, title, content, │
│ deadline, status, priority │
│ ) │
│ status = 'pending' │
└─────────────────────┬────────────────────────────────────┘
▼
┌──────────────────────────────────────────────────────────┐
│ Step 3: 触发推送 │
│ POST /api/wecom/send_message │
│ { │
│ "to_user_id": "xiaowang", │
│ "msg_type": "task_card", │
│ "task_id": "xxx", │
│ "title": "📋 新任务: 完成Q2市场调研报告", │
│ "deadline": "2026-05-20", │
│ "assigner": "张经理", │
│ "actions": [ │
│ {"text": "确认收到", "key": "task_confirm"}, │
│ {"text": "查看详情", "key": "task_detail"} │
│ ] │
│ } │
└─────────────────────┬────────────────────────────────────┘
▼
┌──────────────────────────────────────────────────────────┐
│ 企微推送 → 小王收到任务卡片消息 │
│ ┌──────────────────────────┐ │
│ │ 📋 新任务通知 │ │
│ │ │ │
│ │ 张经理 给你安排了新任务: │ │
│ │ │ │
│ │ 完成Q2市场调研报告 │ │
│ │ │ │
│ │ 截止日期: 2026-05-20 │ │
│ │ │ │
│ │ [确认收到] [查看详情] │ │
│ └──────────────────────────┘ │
└──────────────────────────────────────────────────────────┘
6.4 场景4：管理后端 × 无代码工作流编排
这是整个系统最核心的差异化功能，相当于一个轻量级企业版 Dify。详见下一章。
八、Dify-like 可视化流编排引擎
7.1 核心理念
Dify 的核心: 可视化拖拽 → 生成工作流定义 → 后端引擎执行
我们的实现:
可视化拖拽 → 生成 JSON Flow Definition → Flow Engine → AgentScope Pipeline 执行
关键: AgentScope 的 Pipeline 是 Python 代码级的，
我们需要一个 Flow Engine 作为"JSON → AgentScope"的桥梁。
7.2 流编辑器前端设计
┌──────────────────────────────────────────────────────────────┐
│ 工作流编辑器 [保存] [发布] │
├───────────────┬──────────────────────────────────┬───────────┤
│ │ │ │
│ 节点面板 │ 画布区 │ 属性面板 │
│ │ │ │
│ ┌─────────┐ │ │ 节点名称: │
│ │ 🔔 触发 │ │ ┌──────┐ ┌──────┐ │ LLM处理 │
│ └─────────┘ │ │企微 │───▶│ LLM │ │ │
│ │ │消息 │ │处理 │ │ 模型: │
│ ┌─────────┐ │ └──────┘ └──┬───┘ │ GPT-4o │
│ │ 🤖 LLM │ │ │ │ │
│ └─────────┘ │ ▼ │ Prompt: │
│ │ ┌──────┐ │ "你是... │
│ ┌─────────┐ │ │ MCP │ │ ..." │
│ │ 🔧 工具 │ │ │ 调用 │ │ │
│ └─────────┘ │ └──┬───┘ │ 温度: │
│ │ │ │ 0.7 │
│ ┌─────────┐ │ ▼ │ │
│ │ 📡 MCP │ │ ┌──────┐ │ │
│ └─────────┘ │ │ 企微 │ │ │
│ │ │ 通知 │ │ │
│ ┌─────────┐ │ └──────┘ │ │
│ │ 🔀 条件 │ │ │ │
│ └─────────┘ │ │ │
│ │ │ │
│ ┌─────────┐ │ │ │
│ │ 📚 RAG │ │ │ │
│ └─────────┘ │ │ │
│ │ │ │
│ ┌─────────┐ │ │ │
│ │ 📤 输出 │ │ │ │
│ └─────────┘ │ │ │
│ │ │ │
└───────────────┴──────────────────────────────────┴───────────┘
7.3 流定义 JSON 格式
前端拖拽完成后，生成如下标准 JSON：
{
"id": "flow_doc_processor_v1",
"name": "文档格式修正与导入",
"description": "员工上传文档 → AI修正格式 → 导入系统 → 推送结果",
"version": 1,
"status": "published",
"published_to_wecom": true,
"trigger": {
"type": "wecom_message",
"config": {
"match_type": "keywords",
"keywords": ["文档处理", "格式修正", "导入"],
"accept_files": [".docx", ".xlsx", ".pdf"]
}
},
"nodes": [
{
"id": "node_trigger",
"type": "trigger",
"label": "企微消息触发",
"config": {}
},
{
"id": "node_file_preview",
"type": "tool",
"label": "解析上传文件",
"config": {
"tool_name": "parse_uploaded_file",
"input_mapping": {
"file_url": "{{ trigger.file_url }}",
"file_type": "{{ trigger.file_type }}"
}
}
},
{
"id": "node_llm_format",
"type": "llm",
"label": "AI格式修正",
"config": {
"model": "gpt-4o-mini",
"temperature": 0.3,
"system_prompt": "你是文档格式专家。请检查以下文档内容并按企业标准格式修正。企业格式要求：\n1. 标题使用一级标题\n2. 表格首行加粗\n3. 段落间距统一",
"input_mapping": {
"document_content": "{{ node_file_preview.content }}"
}
}
},
{
"id": "node_mcp_import",
"type": "mcp",
"label": "导入业务系统",
"config": {
"mcp_server": "business_system",
"tool_name": "import_document",
"input_mapping": {
"content": "{{ node_llm_format.output }}",
"author_id": "{{ trigger.user_id }}",
"category": "AUTO_PROCESSED"
}
}
},
{
"id": "node_notify",
"type": "wecom_notify",
"label": "企微通知结果",
"config": {
"target": "{{ trigger.user_id }}",
"message_template": "✅ 文档已处理完成！\n• 文件名: {{ trigger.file_name }}\n• 修正项: {{ node_llm_format.changes }}\n• 已导入系统，可前往查看"
}
}
],
"edges": [
{"from": "node_trigger", "to": "node_file_preview"},
{"from": "node_file_preview", "to": "node_llm_format"},
{"from": "node_llm_format", "to": "node_mcp_import"},
{"from": "node_mcp_import", "to": "node_notify"}
]
}
7.4 流引擎后端实现
引擎的核心：将 JSON 流定义 → 动态创建 AgentScope 组件 → 执行
# backend/modules/flow_engine/engine.py
"""
流引擎核心 - JSON Flow Definition → AgentScope Pipeline 执行
完全不修改 AgentScope 源码，通过创建轻量级的 AgentBase 子类来实现每个节点
"""
from agentscope.agent import AgentBase, ReActAgent
from agentscope.pipeline import sequential_pipeline, MsgHub
from agentscope.message import Msg
from agentscope.tool import Toolkit
class FlowEngine:
"""流引擎：加载 JSON 流定义并动态执行"""
def __init__(self, flow_definition: dict):
self.definition = flow_definition
self.nodes = flow_definition["nodes"]
self.edges = flow_definition["edges"]
self._agent_cache = {} # 节点 → Agent 实例缓存
async def execute(self, input_msg: Msg, context: dict) -> Msg:
"""执行流"""
# 1. 构建执行拓扑 (拓扑排序)
execution_order = self._topological_sort()
# 2. 为每个节点创建 Agent
agents = []
for node_id in execution_order:
agent = await self._create_node_agent(node_id, context)
agents.append(agent)
# 3. 通过 AgentScope SequentialPipeline 串联执行
# 完全不修改 AgentScope，直接使用其能力
result = await sequential_pipeline(agents, msg=input_msg)
return result
async def _create_node_agent(self, node_id: str, context: dict) -> AgentBase:
"""为每个流节点创建一个 Agent，缓存复用"""
if node_id in self._agent_cache:
return self._agent_cache[node_id]
node = self._get_node(node_id)
agent = await NodeAgentFactory.create(node, context)
self._agent_cache[node_id] = agent
return agent
def _topological_sort(self) -> list[str]:
"""拓扑排序，确定执行顺序"""
# ... 基于 edges 计算
pass
class NodeAgentFactory:
"""节点 Agent 工厂 - 根据节点类型创建对应的 Agent 实例"""
@classmethod
async def create(cls, node: dict, context: dict) -> AgentBase:
node_type = node["type"]
config = node["config"]
node_id = node["id"]
if node_type == "trigger":
# 触发节点 = 透传消息，不处理
return PassThroughAgent(node_id)
elif node_type == "llm":
# LLM 节点 → 创建一个轻量 ReActAgent
model = cls._get_model(config["model"])
formatter = cls._get_formatter(config["model"])
return LLMNodeAgent(
name=f"LLM_{node_id}",
sys_prompt=config["system_prompt"],
model=model,
formatter=formatter,
)
elif node_type == "tool":
# 工具节点 → 创建带工具的 Agent
toolkit = Toolkit()
toolkit.register_tool_function(
cls._resolve_tool(config["tool_name"])
)
return ToolNodeAgent(
name=f"Tool_{node_id}",
toolkit=toolkit,
)
elif node_type == "mcp":
# MCP 节点 → AgentScope 原生 MCP 客户端
return MCPNodeAgent(
name=f"MCP_{node_id}",
mcp_client=cls._create_mcp_client(config),
)
elif node_type == "rag":
# RAG 节点 → AgentScope 原生 KnowledgeBase
return RAGNodeAgent(
name=f"RAG_{node_id}",
knowledge_base=await cls._create_kb(config),
)
elif node_type == "condition":
# 条件分支节点 → 特殊 Agent，根据 LLM 判断走哪个分支
return ConditionNodeAgent(
name=f"Cond_{node_id}",
condition=config["condition"],
true_agent=await cls.create({"id": config["true_branch"]}, context),
false_agent=await cls.create({"id": config["false_branch"]}, context),
)
elif node_type == "wecom_notify":
return WeComNotifyAgent(node_id, config)
else:
raise ValueError(f"Unknown node type: {node_type}")
节点 Agent 的轻量实现（全部继承 AgentBase，不修改源码）：
# backend/agentscope_integration/agents/flow_node_agent.py
from agentscope.agent import AgentBase
from agentscope.message import Msg, TextBlock
from agentscope.tool import Toolkit
class PassThroughAgent(AgentBase):
"""透传节点 - 不处理消息，直接传递给下一个"""
async def reply(self, msg, **kwargs) -> Msg:
return msg # 原样返回
async def observe(self, msg) -> None:
pass
class LLMNodeAgent(AgentBase):
"""
LLM 处理节点。
每个流中的 LLM 节点就是一个简化的 Agent，只做一轮 LLM 调用。
"""
def __init__(self, name, sys_prompt, model, formatter):
super().__init__()
self.name = name
self._sys_prompt = sys_prompt
self.model = model
self.formatter = formatter
async def reply(self, msg: Msg, **kwargs) -> Msg:
user_text = msg.get_text_content()
prompt = await self.formatter.format([
Msg("system", self._sys_prompt, "system"),
Msg("user", user_text, "user"),
])
res = await self.model(prompt)
res_text = ""
if isinstance(res, list):
res_text = res[0].get_text_content()
else:
res_text = res.get_text_content()
return Msg(self.name, res_text, "assistant")
async def observe(self, msg) -> None:
pass
class ToolNodeAgent(AgentBase):
"""工具执行节点"""
def __init__(self, name, toolkit: Toolkit):
super().__init__()
self.name = name
self.toolkit = toolkit
async def reply(self, msg: Msg, **kwargs) -> Msg:
# 解析消息中的工具调用意图，执行对应工具
# 简化实现：取消息文本作为工具参数
pass
async def observe(self, msg) -> None:
pass
7.5 流的上架机制
流在编辑器保存后 → status: draft
管理员点击 "上架到企微"
│
▼
┌──────────────────────────────────┐
│ 1. 流定义验证 │
│ · JSON Schema 校验 │
│ · 节点拓扑完整性检查 │
│ · MCP 服务连通性测试 │
└──────────────┬───────────────────┘
▼
┌──────────────────────────────────┐
│ 2. 注册到企微消息路由 │
│ · 前端配置的关键词/文件类型 │
│ 被注册为触发规则 │
│ · status → published │
└──────────────┬───────────────────┘
▼
┌──────────────────────────────────┐
│ 3. 热部署 (无需重启) │
│ · Flow Engine 内存中加载新定义 │
│ · 下次匹配到的消息将走新流 │
└──────────────────────────────────┘
7.6 流引擎与 AgentScope 的衔接点总结
┌──────────────────────────────────────────────────────────────┐
│ 流引擎 (我们写的) AgentScope (不改源码) │
│ │
│ FlowEditor (前端拖拽) │
│ │ │
│ ▼ │
│ JSON Flow Definition │
│ │ │
│ ▼ │
│ FlowEngine.execute() │
│ │ │
│ ├── NodeAgentFactory → PassThroughAgent 继承 AgentBase│
│ ├── NodeAgentFactory → LLMNodeAgent 继承 AgentBase│
│ ├── NodeAgentFactory → ToolNodeAgent 使用 Toolkit │
│ ├── NodeAgentFactory → MCPNodeAgent 使用 MCPClient│
│ ├── NodeAgentFactory → RAGNodeAgent 使用 Knowledge│
│ │ │
│ ▼ │
│ sequential_pipeline([agents...], msg) ←──────── 直接用! │
│ │ │
│ ▼ │
│ 输出 Msg → 转换为企微消息 │
└──────────────────────────────────────────────────────────────┘
九、数据库设计
8.1 数据库选型
| 数据库 | 用途 | Docker 镜像 |
|---|---|---|
| PostgreSQL 16 | 主业务数据库（用户/部门/角色/权限/任务/流定义等） | postgres:16-alpine |
| Redis 7 | 缓存（会话缓存/Agent实例缓存/任务队列/限流） | redis:7-alpine |
| Qdrant | 向量数据库（RAG文档嵌入/长期记忆向量检索） | qdrant/qdrant:latest |
| MinIO | 对象存储（员工上传的文件/文档/图片） | minio/minio:latest |
8.2 完整 ER 关系
departments ──┬── users ──┬── user_roles ──── roles ──┬── role_permissions ─── permissions
(树形) │ │ │
│ │ └── role_data_scopes ─── data_scopes
│ │
│ ├── tasks (assigner_id, assignee_id)
│ │
│ ├── chat_sessions (user_id, agent_type)
│ │
│ └── audit_logs (operator_id)
│
└── flow_definitions ─── flow_versions
(creator: users)
权限继承规则:
· 上级自动继承下级的可见权限
· 部门经理默认可以看到本部门全部数据
· 跨部门绝对隔离，通过 org_path 字段实现
8.3 200人规模下的分库策略
对于200人公司，单 PostgreSQL 实例完全够用，无需分库。但逻辑上建议按以下 schema 组织：
-- Schema 分层
CREATE SCHEMA core; -- 核心业务: users, departments, roles, permissions
CREATE SCHEMA workflow; -- 工作流: flow_definitions, flow_versions, flow_executions
CREATE SCHEMA task; -- 任务: tasks, task_logs
CREATE SCHEMA agent; -- 智能体: chat_sessions, agent_configs
CREATE SCHEMA audit; -- 审计: audit_logs, access_logs
十、Docker 部署方案
9.1 容器拓扑
┌──────────────────────────────────────────────────────────┐
│ Docker Network: enterprise-net │
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ nginx │ │ frontend │ │ backend │ │
│ │ (gateway) │ │ (Vue SPA) │ │ (FastAPI) │ │
│ │ port: 80/443 │ │ port: 3000 │ │ port: 8000 │ │
│ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │
│ │ │ │ │
│ └──────────────────┼──────────────────┘ │
│ │ │
│ ┌───────┼───────┐ │
│ │ │ │ │
│ ┌─────▼──┐ ┌─▼──────┐ ┌▼──────────┐ │
│ │postgres│ │ redis │ │ qdrant │ │
│ │ :5432 │ │ :6379 │ │ :6333 │ │
│ └────────┘ └────────┘ └────────────┘ │
│ │
│ ┌──────────┐ │
│ │ minio │ │
│ │ :9000 │ ← 文件存储 │
│ └──────────┘ │
└──────────────────────────────────────────────────────────┘
9.2 Docker Compose 配置
# docker-compose.yml
version: "3.8"
services:
# ==================== 基础设施 ====================
postgres:
image: postgres:16-alpine
container_name: ent-postgres
restart: always
environment:
POSTGRES_USER: enterprise
POSTGRES_PASSWORD: ${DB_PASSWORD:-enterprise123}
POSTGRES_DB: enterprise_ai
volumes:
- postgres_data:/var/lib/postgresql/data
- ./init-db:/docker-entrypoint-initdb.d # 初始化脚本
ports:
- "5432:5432"
healthcheck:
test: ["CMD-SHELL", "pg_isready -U enterprise"]
interval: 5s
timeout: 5s
retries: 5
redis:
image: redis:7-alpine
container_name: ent-redis
restart: always
command: redis-server --appendonly yes --requirepass ${REDIS_PASSWORD:-redis123}
volumes:
- redis_data:/data
ports:
- "6379:6379"
qdrant:
image: qdrant/qdrant:latest
container_name: ent-qdrant
restart: always
volumes:
- qdrant_data:/qdrant/storage
ports:
- "6333:6333"
- "6334:6334"
minio:
image: minio/minio:latest
container_name: ent-minio
restart: always
command: server /data --console-address ":9001"
environment:
MINIO_ROOT_USER: ${MINIO_USER:-minioadmin}
MINIO_ROOT_PASSWORD: ${MINIO_PASSWORD:-minioadmin123}
volumes:
- minio_data:/data
ports:
- "9000:9000"
- "9001:9001"
# ==================== 后端服务 ====================
backend:
build:
context: ./backend
dockerfile: Dockerfile
container_name: ent-backend
restart: always
depends_on:
postgres:
condition: service_healthy
redis:
condition: service_started
qdrant:
condition: service_started
minio:
condition: service_started
environment:
DATABASE_URL: postgresql+asyncpg://enterprise:${DB_PASSWORD:-enterprise123}@postgres:5432/enterprise_ai
REDIS_URL: redis://:${REDIS_PASSWORD:-redis123}@redis:6379
QDRANT_URL: http://qdrant:6333
MINIO_URL: http://minio:9000
MINIO_ACCESS_KEY: ${MINIO_USER:-minioadmin}
MINIO_SECRET_KEY: ${MINIO_PASSWORD:-minioadmin123}
SECRET_KEY: ${JWT_SECRET:-your-super-secret-key}
LLM_API_KEY: ${LLM_API_KEY}
LLM_API_BASE: ${LLM_API_BASE:-https://api.openai.com/v1}
WECOM_CORP_ID: ${WECOM_CORP_ID}
WECOM_SECRET: ${WECOM_SECRET}
WECOM_TOKEN: ${WECOM_TOKEN}
WECOM_AES_KEY: ${WECOM_AES_KEY}
AGENTSCOPE_RUNTIME_MODE: daemon_thread # Runtime 运行模式
ports:
- "8000:8000" # AgentApp HTTP + AG-UI
volumes:
- ./backend:/app # 开发环境挂载，生产去掉
# ==================== 前端 ====================
frontend:
build:
context: ./frontend
dockerfile: Dockerfile
container_name: ent-frontend
restart: always
depends_on:
- backend
ports:
- "3000:80"
# ==================== 网关 ====================
nginx:
image: nginx:alpine
container_name: ent-nginx
restart: always
depends_on:
- frontend
- backend
volumes:
- ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
- ./nginx/ssl:/etc/nginx/ssl:ro # HTTPS证书
ports:
- "80:80"
- "443:443"
volumes:
postgres_data:
redis_data:
qdrant_data:
minio_data:
9.3 Backend Dockerfile
# backend/Dockerfile
FROM python:3.11-slim
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
build-essential \
libpq-dev \
&& rm -rf /var/lib/apt/lists/*
# 安装 Python 依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# requirements.txt 内容:
# fastapi>=0.110.0
# uvicorn>=0.27.0
# sqlalchemy[asyncio]>=2.0
# asyncpg>=0.29.0
# redis>=5.0
# python-jose[cryptography]>=3.3.0
# passlib[bcrypt]>=1.7.4
# python-multipart>=0.0.9
# httpx>=0.27.0
# minio>=7.2.0
# qdrant-client>=1.7.0
# agentscope>=1.0.19 ← AgentScope 作为依赖安装
# agentscope[full]>=1.0.19 ← 完整功能
# agentscope-runtime>=1.1.0 ← AgentScope Runtime 网关层 (AgentApp IS FastAPI)
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
9.4 Frontend Dockerfile
# frontend/Dockerfile (多阶段构建)
FROM node:20-alpine AS build
WORK /app
COPY package*.json ./
RUN npm ci
COPY . .
RUN npm run build
FROM nginx:alpine
COPY --from=build /app/dist /usr/share/nginx/html
COPY nginx-frontend.conf /etc/nginx/conf.d/default.conf
EXPOSE 80
9.5 Nginx 网关配置
# nginx/nginx.conf
upstream backend_api {
server backend:8000;
}
upstream frontend_app {
server frontend:80;
}
server {
listen 80;
server_name your-domain.com;
# 管理后台前端
location / {
proxy_pass http://frontend_app;
}
# API 转发
location /api/ {
proxy_pass http://backend_api/api/;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
}
# 企业微信回调（需要公网可达）
location /wecom/ {
proxy_pass http://backend_api/wecom/;
}
# WebSocket
location /ws/ {
proxy_pass http://backend_api/ws/;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
}
}
十一、API 接口设计
10.1 RESTful API 总览
认证相关:
POST /api/auth/login 用户登录
POST /api/auth/refresh 刷新Token
GET /api/auth/me 获取当前用户信息
组织架构:
GET /api/org/departments 部门树
POST /api/org/departments 创建部门
PUT /api/org/departments/{id} 更新部门
DELETE /api/org/departments/{id} 删除部门
GET /api/org/users 用户列表 (受RBAC过滤)
GET /api/org/users/{id} 用户详情
PUT /api/org/users/{id} 更新用户
角色权限:
GET /api/roles 角色列表
POST /api/roles 创建角色
PUT /api/roles/{id}/permissions 配置功能权限
PUT /api/roles/{id}/data-scope 配置数据范围
工作监控:
GET /api/monitor/subordinates 获取可查看的下属列表
GET /api/monitor/employee/{id}/dashboard 员工工作看板
GET /api/monitor/employee/{id}/analysis AI分析报告
任务管理:
POST /api/tasks 创建任务
GET /api/tasks 任务列表 (按RBAC过滤)
GET /api/tasks/{id} 任务详情
PUT /api/tasks/{id} 更新任务状态
POST /api/tasks/{id}/push 推送任务到企微
流编排:
GET /api/flows 流列表
POST /api/flows 创建流定义
GET /api/flows/{id} 流详情 (含JSON定义)
PUT /api/flows/{id} 更新流定义
POST /api/flows/{id}/publish 发布/上架流到企微
POST /api/flows/{id}/unpublish 下架
POST /api/flows/{id}/test 测试执行流
GET /api/flows/market 已上架的流市场
智能体:
GET /api/agents 智能体列表
POST /api/agents/{id}/chat 与指定智能体对话 (测试)
企微配置:
GET /api/wecom/config 企微配置
PUT /api/wecom/config 更新企微配置
POST /api/wecom/callback 企微回调 (公网)
MCP服务:
GET /api/mcp/servers MCP服务列表
POST /api/mcp/servers 注册MCP服务
PUT /api/mcp/servers/{id} 更新MCP配置
DELETE /api/mcp/servers/{id} 注销MCP服务
POST /api/mcp/servers/{id}/test 测试MCP连接
审计日志:
GET /api/audit/logs 审计日志 (按RBAC过滤)
10.2 核心 API 示例
# POST /api/tasks - 创建任务
{
"title": "完成Q2市场调研报告",
"content": "请针对竞品A、B、C进行市场分析，输出调研报告。",
"assignee_id": "uuid-of-xiaowang",
"deadline": "2026-05-20T18:00:00Z",
"priority": "high",
"push_to_wecom": true
}
# Response:
{
"code": 200,
"data": {
"task_id": "uuid",
"status": "pending",
"wecom_message_sent": true,
"wecom_message_id": "msg_xxx"
}
}
# GET /api/monitor/employee/{id}/analysis
# Response:
{
"code": 200,
"data": {
"employee": {
"id": "uuid",
"name": "小王",
"department": "市场部"
},
"period": "2026-05-01 ~ 2026-05-09",
"analysis": {
"task_completion_rate": 0.85,
"active_days": 7,
"total_interactions": 42,
"main_topics": ["市场调研", "竞品分析", "文档处理", "数据统计"],
"efficiency_trend": "提升",
"efficiency_detail": "本周文档处理效率比上周提升23%，主要因为熟练使用AI格式修正功能",
"strengths": ["数据分析能力强", "主动使用AI工具", "文档规范意识好"],
"growth_suggestions": ["建议学习进阶SQL分析工具", "可参与跨部门数据分析项目"],
"personality_traits": "工作认真细致，善于利用工具提升效率，有一定数据分析天赋"
}
}
}
十二、开发路线图
Phase 1: 基础设施 + Runtime + 企微接入 (4周) ✅ 已完成
目标: 跑通 AgentScope Runtime → AgentScope → 企微消息回路的完整链路
Week 1-2:
✅ Docker Compose 环境搭建 (PostgreSQL + Redis + Nginx + backend + frontend 五容器)
✅ 数据库表创建 (12张表: departments, users, roles, permissions 等)
✅ 用户系统基础 CRUD (user/role/permission)
✅ JWT 认证中间件 + RBAC中间件 (双中间件挂载到 FastAPI)
✅ AgentScope Runtime 集成验证
Week 3-4:
✅ 企业微信回调路由 (消息接收/回复)
✅ 企微消息推送封装 (真实API调用逻辑)
✅ AgentScope 集成层搭建 (Agent工厂 + UserIsolatedMemory + RBAC钩子)
✅ 员工AI助手MVP (EmployeeAI, ManagerAI, TaskAI, DocumentAI)
✅ 工具注册 (wecom_tools + document_tools)
交付物:
· 4类AI Agent就绪
· 企微消息回路可用
· RBAC中间件生效
· 记忆隔离保证数据安全
Phase 2: 管理后台 + 双RBAC (4周) ✅ 已完成
目标: 管理后台可用，权限体系就绪
Week 5-6:
✅ 管理后台前端项目初始化 (Vue3 + Element Plus + Vite + Pinia)
✅ 登录/仪表盘/组织架构页面
✅ 双RBAC引擎开发 (平台RBAC: 功能权限控制; 组织RBAC: 数据可见范围)
✅ 权限中间件集成 (router beforeEach + API middleware + 侧边栏条件渲染)
Week 7-8:
✅ 员工工作监控页面 (员工列表 + 工作看板)
✅ LLM分析报告功能 (调用 AgentScope OpenAIChatModel 生成结构化分析)
✅ 任务创建与推送功能 (CRUD + 企微推送)
✅ 智能体管理页面 (4类Agent + 对话界面)
交付物:
· 前端13个页面就绪
· 双RBAC贯穿前后端
· 下属工作监控可用
· 任务分派+企微推送可用
Phase 3: 流编排引擎 (6周) ✅ 已完成
目标: Dify-like 可视化流编排可用
Week 9-11:
✅ 流编辑器前端 (Vue3拖拽画布: 节点面板+连线+属性配置)
✅ 流JSON定义格式确定 (nodes + edges + trigger)
✅ Flow Engine 核心 (JSON → 拓扑排序 → 动态创建Agent → Pipeline执行)
✅ 节点类型: trigger, llm, tool, output
Week 12-14:
✅ 节点类型: mcp, condition, rag, wecom_notify (7种节点全部实现)
✅ 流发布/上架机制 (publish/unpublish + wecom可用标记)
✅ 流测试功能 (API验证: 连通性/重名/缺失触发)
✅ 流市场页面 (已上架流展示)
交付物:
· 可视化拖拽流编辑器可用
· Flow Engine 可执行JSON定义
· 上架/下架/测试/执行完整API
Phase 4: 高级功能 + 完善 (4周) ✅ 已完成
Week 15-18:
✅ MCP 服务注册中心 (DB存储 + 连接测试 + 工具发现 + 审计日志)
✅ 文档处理服务 (文件上传/解析/格式修正/删除 + 审计日志)
✅ 审计日志完整功能 (分页筛选/操作统计/CSV导出)
✅ 通知系统完善 (WebSocket实时推送 + 企微通道 + 通知模板管理)
✅ 速率限制中间件 (Token Bucket 算法, 429保护)
✅ Redis缓存系统 (双通道：Redis优先 → 本地内存降级)
✅ 连接池优化 (pool_size=20, max_overflow=40, pool_pre_ping)
✅ 系统监控面板 (健康检查/使用统计/缓存状态/CPU内存指标)
交付物:
· 完整的企业级 AI 平台核心功能
· 14个后端API模块 + 20个前端页面
Phase 5: WebSocket 实时通知 + 模板系统 ✅ 已完成
✅ WebSocket 管理器 (多用户连接管理/心跳/广播/单播)
✅ 企微真实API推送集成 (access_token获取 + 消息推送)
✅ 通知模板CRUD (多通道: wecom/in-app)
✅ 系统模板保护 (is_system 字段防删)
✅ WebSocket连接统计端点
交付物:
· 实时通知频道可用
· 前端通知中心页面 (WebSocket连接 + 发送 + 模板管理)
Phase 6: 安全加固 + 性能优化 ✅ 已完成
✅ 速率限制中间件 (全API保护，白名单: /health, /auth/login)
✅ 输入校验增强 (文件大小限制: MAX_UPLOAD_SIZE_MB)
✅ JWT认证依赖注入 (get_current_user + require_permission)
✅ 数据库连接池优化 (pool_size/max_overflow/pool_pre_ping/pool_recycle)
✅ Redis缓存层 (角色/权限/Agent/会话缓存，TTL自动过期)
✅ 本地内存缓存降级 (Redis不可用时自动切换)
✅ 缓存清除管理端点
交付物:
· API速率保护就绪
· 连接池优化完成
· 缓存双通道策略
Phase 7: 系统监控 + 前端高级功能 ✅ 已完成
✅ 系统健康检查端点 (/api/system/health: DB/Redis/CPU/内存/运行时间)
✅ 使用统计端点 (/api/system/stats: 用户/会话/消息/任务/流/API调用量)
✅ 指标收集端点 (SystemMetric DB模型 + API上报)
✅ 缓存管理 (清除/查看缓存状态)
✅ 速率限制统计端点
✅ 前端系统监控页面 (实时指标 + 统计卡片 + 缓存控制)
✅ 前端文档管理页面 (文件上传/解析/预览/格式化)
✅ 前端通知中心页面 (WebSocket连接/消息/模板)
✅ 前端审计日志增强 (筛选/分页/统计/CSV导出)
✅ 前端图表依赖引入 (echarts + vue-echarts)
交付物:
· 系统运维监控面板可用
· 文档管理前后端完整闭环
· 通知中心前后端完整闭环
· 审计日志支持筛选导出
十三、项目目录结构
实际项目结构 (enterprise-platform/)
enterprise-platform/ # 位于 agentscope 目录下
│
├── docker-compose.yml # 五容器一键部署
├── .env # 环境变量 (可选)
├── .env.example # 环境变量模板
│
├── nginx/ # 统一入口 Nginx
│ └── nginx.conf # 反向代理 (前端SPA + 后端API)
│
├── init-db/ # PostgreSQL 初始化SQL
│ └── 01-init.sql # 15张表 + 种子数据 (admin/admin123)
│
├── backend/ # 后端 (FastAPI，可升级为 AgentApp)
│ ├── Dockerfile
│ ├── requirements.txt # fastapi + sqlalchemy + redis + agentscope
│ ├── main.py # FastAPI 入口 (14个路由模块)
│ ├── config.py # pydantic-settings 配置
│ ├── database.py # 异步引擎 + 连接池
│ ├── dependencies.py # get_current_user + require_permission
│ │
│ ├── modules/ # 14个业务模块
│ │ ├── auth/ # JWT 认证
│ │ ├── org/ # 部门 + 人员管理
│ │ ├── rbac/ # 双RBAC (平台+组织)
│ │ ├── wecom/ # 企微消息回调 + 推送
│ │ ├── agent_manager/ # 智能体管理 + 对话
│ │ ├── task/ # 任务CRUD + 企微推送
│ │ ├── monitor/ # 员工监控 + LLM分析
│ │ ├── mcp_registry/ # MCP服务注册中心
│ │ ├── flow_engine/ # 流编排引擎 (Dify-like)
│ │ │ ├── engine.py # Flow Engine 核心
│ │ │ └── router.py # 流CRUD + 执行API
│ │ ├── audit/ # 审计日志 (分页/筛选/导出)
│ │ ├── document/ # 文档管理 (上传/解析/格式化)
│ │ ├── notification/ # WebSocket 实时通知 + 模板
│ │ └── system/ # 系统监控 (健康/统计/缓存)
│ │
│ ├── agentscope_integration/ # AgentScope 集成层
│ │ ├── factory.py # 4种Agent工厂
│ │ ├── hooks/rbac_hook.py # RBAC上下文钩子
│ │ ├── memory/user_memory.py # 用户隔离记忆
│ │ └── tools/ # 企微/文档工具
│ │
│ ├── models/__init__.py # 15个SQLAlchemy模型
│ ├── schemas/__init__.py # Pydantic 模型
│ └── middleware/ # 4个中间件
│ ├── rbac_middleware.py # JWT + RBAC
│ ├── rate_limiter.py # Token Bucket 速率限制
│ └── cache_manager.py # Redis/本地双通道缓存
│
├── frontend/ # Vue 3 管理后台 (20个页面)
│ ├── Dockerfile
│ ├── package.json
│ ├── vite.config.ts
│ ├── nginx.conf # SPA静态服务 + API代理 + WS支持
│ ├── index.html
│ └── src/
│ ├── main.ts # Element Plus + Pinia + Router
│ ├── App.vue
│ ├── router/index.ts # 路由 + 权限守卫
│ ├── stores/user.ts # 用户状态 + 权限
│ ├── api/index.ts # axios 封装 (14组API)
│ ├── components/layout/
│ │ └── MainLayout.vue # 侧边栏 + 顶栏布局
│ └── views/
│ ├── login/Login.vue
│ ├── dashboard/Dashboard.vue
│ ├── org/ # DepartmentTree + UserList
│ ├── role/ # RoleList + PermissionConfig
│ ├── monitor/ # EmployeeList + WorkDashboard + AIAnalysis
│ ├── task/ # TaskList + TaskCreate + TaskDetail
│ ├── flow/ # FlowList + FlowEditor + FlowMarket
│ ├── wecom/BotConfig.vue
│ ├── agent/ # AgentList + AgentChat
│ ├── audit/AuditLog.vue # 筛选/分页/统计/导出
│ ├── document/DocumentManager.vue # 上传/解析/格式化
│ ├── notification/NotificationCenter.vue # WebSocket + 模板
│ └── system/SystemMonitor.vue # 健康/统计/缓存
│
├── docs/ # 文档 (待完善)
│
└── scripts/ENTERPRISE_PLAN.md # 本计划文档
与 AgentScope + Runtime 的关系：本项目当前使用纯 FastAPI（非 AgentApp），通过 pip 依赖
agentscope。AgentScope 提供 Agent 引擎（ReActAgent、Memory、Pipeline、Tools、Hooks），本项目通过继承/钩子/封装方式使用，不修改 AgentScope 任何源码。Future：可升级为
AgentApp(FastAPI)以获取 AG-UI / A2A 协议 / 原生 MCP 能力，仅需更改main.py中的FastAPI()→AgentApp()并注册现有路由为 custom_endpoints。
附录A：项目启动指南
前置条件
# Docker Desktop (Windows/Mac/Linux)
# 确保 Docker Compose V2 可用
docker compose version
快速启动 (Docker 一键部署)
cd enterprise-platform
# 1. 配置API Key (可选，有测试用默认值)
cp .env.example .env
# 编辑 .env 填入:
# LLM_API_KEY=sk-your-real-api-key
# LLM_API_BASE=https://api.openai.com/v1
# LLM_MODEL=gpt-4o-mini
# 2. 启动全部服务 (5个容器)
docker compose up -d
# 3. 等待初始化 (首次启动需下载镜像 + 初始化DB约2-3分钟)
docker compose logs -f backend # 看到 "Application startup complete" 即就绪
# 4. 访问
# 管理后台: http://localhost:80
# API文档: http://localhost:8000/docs
# 健康检查: http://localhost:8000/health
分步启动 (开发模式)
# 1. 启动基础设施
docker compose up -d postgres redis
# 2. 启动后端 (命令行，方便调试)
cd backend
pip install -r requirements.txt
uvicorn main:app --reload --port 8000
# 3. 启动前端 (另一个终端)
cd frontend
npm install
npm run dev # http://localhost:3000，自动代理到 :8000
默认账号
| 账号 | 密码 | 角色 |
|---|---|---|
admin |
admin123 |
超级管理员 (全部权限) |
容器架构
Browser:80 → nginx → / → frontend:80 (Vue SPA)
/api/* → backend:8000 (FastAPI)
/wecom/* → backend:8000
← WebSocket /api/notification/ws/* → backend:8000
如何切换为 AgentApp (可选)
当前使用 纯 FastAPI，如需 AgentScope Runtime 能力：
# main.py - 将 FastAPI 替换为 AgentApp
from agentscope_runtime import AgentApp
app = AgentApp(
config_file="config.toml", # AgentScope 配置
custom_endpoints=[(router, "") for router in all_routers], # 注册现有路由
)
附录B：AgentScope 零修改验证清单
| 功能 | 使用的 AgentScope 能力 | 修改源码? | 验证方式 |
|---|---|---|---|
| 员工AI助手 | 继承 ReActAgent | ❌ 否 | 子类化 + reply() 调用 |
| 管理分析助手 | 继承 ReActAgent | ❌ 否 | 子类化 + reply() 调用 |
| 任务管理助手 | 继承 ReActAgent | ❌ 否 | 子类化 + reply() 调用 |
| 文档处理助手 | 继承 ReActAgent | ❌ 否 | 子类化 + reply() 调用 |
| RBAC 上下文注入 | AgentScope hooks (pre_reply) | ❌ 否 | register_instance_hook |
| 记忆隔离 | 继承 MemoryBase 封装 UserIsolatedMemory | ❌ 否 | get_memory() 过滤测试 |
| 自定义工具 | Toolkit + ToolFunction 注册 | ❌ 否 | 企微/文档工具端到端测试 |
| LLM 调用 (监控分析) | OpenAIChatModel 直接调用 | ❌ 否 | 分析报告生成验证 |
| Flow Engine 节点 | 继承 AgentBase 创建节点Agent | ❌ 否 | Flow 执行验证 |
| 所有功能 | 子类化/钩子/封装/组合模式 | ❌ 否 | AgentScope 源码零修改 |
当前已使用 AgentScope Runtime 的 AgentApp 作为网关。主入口
main.py由FastAPI()迁移至AgentApp()，14个模块路由通过include_router()注册，中间件通过app.middleware("http")挂载。AgentApp 内置 CORS、健康检查、SSE 流式处理、中断管理等能力。
结论: AgentScope 架构设计优秀，通过 Python 标准继承（ReActAgent/AgentBase/MemoryBase）和 AgentApp 原生机制（include_router/中间件/Depends）即可构建完整企业级平台，不需要修改任何源码。