Enterprise 层技术实现

文档说明

本文档描述 Enterprise 层的技术实现细节，包括认证、审计、多租户等模块的设计和接口。

相关文档：

组织级概览：整体架构和业务场景
Scope 设计：知识可见性和权限控制
知识管理：知识结构和检索机制

概述

Enterprise 层提供组织级功能，包括认证、审计、多租户支持等。单向依赖 Core 层，通过扩展点集成。

架构分层

┌─────────────────────────────────────────────────┐
│  Integrations Layer (integrations/)             │
│  - 第三方平台集成（飞书、Slack 等）               │
│  - 依赖：Core + Enterprise                       │
└────────────────┬────────────────────────────────┘
                 │
                 ▼
┌─────────────────────────────────────────────────┐
│  Enterprise Layer (enterprise/)                 │
│  - 组织功能（认证、审计、多租户）                  │
│  - 依赖：Core                                    │
└────────────────┬────────────────────────────────┘
                 │
                 ▼
┌─────────────────────────────────────────────────┐
│  Core Layer (agent/)                            │
│  - 核心 Agent 能力                               │
│  - 零外部依赖                                    │
└─────────────────────────────────────────────────┘

模块结构

enterprise/
├── gateway/                    # 网关层
│   ├── middleware.py           # 中间件基类
│   ├── auth.py                 # 认证中间件
│   ├── audit.py                # 审计中间件
│   ├── rate_limit.py           # 限额中间件
│   └── router.py               # 企业级路由
│
├── auth/                       # 认证系统
│   ├── providers/              # 认证提供者
│   │   ├── api_key.py          # API Key 认证
│   │   ├── jwt.py              # JWT 认证
│   │   └── oauth.py            # OAuth 认证
│   ├── models.py               # User, APIKey, Role
│   └── permissions.py          # 权限检查
│
├── audit/                      # 审计系统
│   ├── logger.py               # 审计日志记录器
│   ├── events.py               # 事件定义
│   ├── storage.py              # 存储接口
│   └── query.py                # 日志查询
│
└── multi_tenant/               # 多租户支持
    ├── tenant.py               # 租户管理
    ├── isolation.py            # 数据隔离
    └── quota.py                # 配额管理

Core 层扩展点

1. 中间件扩展点

位置：agent/trace/api.py:TraceAPI.__init__

class TraceAPI:
    def __init__(
        self,
        trace_store: TraceStore,
        runner: Optional[AgentRunner] = None,
        middlewares: Optional[List[Callable]] = None  # 扩展点
    )

用途：在 API 入口注入认证、审计、限额等中间件

Enterprise 实现：

enterprise/gateway/auth.py:auth_middleware
enterprise/gateway/audit.py:audit_middleware
enterprise/gateway/rate_limit.py:rate_limit_middleware

2. 钩子扩展点

位置：agent/core/runner.py:AgentRunner.__init__

class AgentRunner:
    def __init__(
        self,
        trace_store: TraceStore,
        hooks: Optional[Dict[str, List[Callable]]] = None  # 扩展点
    )

钩子类型：

trace.created：Trace 创建后
llm.called：LLM 调用后
trace.completed：Trace 完成后
tool.executed：工具执行后

Enterprise 实现：

enterprise/audit/hooks.py:audit_trace_created
enterprise/audit/hooks.py:audit_llm_called
enterprise/audit/hooks.py:audit_trace_completed

3. Context 注入

位置：agent/core/runner.py:RunConfig.context

@dataclass
class RunConfig:
    context: Optional[Dict[str, Any]] = None  # 扩展点

用途：传递用户身份、权限、来源等信息

Context 结构：

{
    "user_id": "user_123",
    "org_id": "company",
    "role": "member",
    "source": "feishu",
    "ip": "192.168.1.100"
}

认证系统

API Key 认证

格式：ak_{org_id}_{user_id}_{random}

示例：ak_company_user123_a1b2c3d4

数据模型：

CREATE TABLE api_keys (
    id TEXT PRIMARY KEY,
    key_hash TEXT UNIQUE,
    user_id TEXT REFERENCES users(id),
    org_id TEXT,
    name TEXT,
    scopes JSON,
    rate_limit INTEGER,
    expires_at TIMESTAMP,
    last_used_at TIMESTAMP,
    created_at TIMESTAMP
);

实现位置：enterprise/auth/providers/api_key.py

飞书 OAuth

流程：

飞书 Webhook 携带 user_open_id
查询用户映射表
构建执行上下文

数据模型：

CREATE TABLE feishu_users (
    feishu_user_id TEXT PRIMARY KEY,
    system_user_id TEXT REFERENCES users(id),
    org_id TEXT,
    department_id TEXT,
    name TEXT,
    created_at TIMESTAMP
);

实现位置：integrations/feishu/auth.py

权限系统

角色定义

角色	权限
admin	所有权限
member	查询代码库、创建 Trace、读写知识
viewer	只读权限

权限检查点

API 入口：验证身份和基础权限
资源访问：检查 scope 权限（知识访问）
操作执行：检查操作权限（创建、删除等）

实现位置：enterprise/auth/permissions.py

审计系统

审计日志类型

类型	文件	内容
操作审计	audit.jsonl	API 调用、资源访问、配置修改
成本审计	cost.jsonl	Token 消耗、API 调用成本
安全审计	security.jsonl	认证失败、权限拒绝、异常访问
知识审计	knowledge.jsonl	知识的创建、修改、查询

操作审计格式

{
  "id": "audit_001",
  "timestamp": "2026-03-04T02:30:00Z",
  "user_id": "user_123",
  "org_id": "company",
  "action": "trace.create",
  "resource_type": "trace",
  "resource_id": "trace-abc-123",
  "status": "success",
  "tokens_used": 1500,
  "cost": 0.015,
  "metadata": {}
}

审计记录时机

auth.*：认证相关事件
trace.*：Trace 生命周期事件
knowledge.*：知识操作事件
permission.denied：权限拒绝事件

实现位置：

enterprise/audit/logger.py：日志记录器
enterprise/audit/events.py：事件定义
enterprise/audit/storage.py：存储接口

限额控制

限额维度

用户级：

每日 API 调用次数：1000
每日 Token 消耗：100K
并发 Trace 数：3

组织级：

每月总 Token 消耗：10M
存储空间：100GB
知识条目数：10K

检查时机

API 入口：检查调用次数
Trace 创建前：检查并发数和 Token 配额
知识保存前：检查存储空间