架构师AI杜

Appearance

第85天：企业知识库-管理后台与部署 ​

学习目标 ​

  • 掌握管理后台开发
  • 学习文档管理功能
  • 理解用户管理功能
  • 掌握部署配置
  • 学习监控告警

管理后台开发 ​

后台架构 ​

python
from fastapi import FastAPI, Depends, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import OAuth2PasswordBearer
from typing import Dict, List, Optional

app = FastAPI(title="企业知识库管理后台")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class AdminBackend:
    def __init__(
        self,
        document_storage,
        vector_store,
        user_manager
    ):
        self.document_storage = document_storage
        self.vector_store = vector_store
        self.user_manager = user_manager
    
    async def get_dashboard_stats(self) -> Dict:
        total_documents = await self._get_total_documents()
        total_users = await self._get_total_users()
        total_queries = await self._get_total_queries()
        storage_usage = await self._get_storage_usage()
        
        return {
            "total_documents": total_documents,
            "total_users": total_users,
            "total_queries": total_queries,
            "storage_usage": storage_usage
        }
    
    async def _get_total_documents(self) -> int:
        return 0
    
    async def _get_total_users(self) -> int:
        return 0
    
    async def _get_total_queries(self) -> int:
        return 0
    
    async def _get_storage_usage(self) -> Dict:
        return {
            "total": 0,
            "used": 0,
            "available": 0
        }

文档管理API ​

python
from fastapi import APIRouter, UploadFile, File, Query
from pydantic import BaseModel, Field

router = APIRouter(prefix="/api/admin/documents", tags=["文档管理"])

class DocumentListResponse(BaseModel):
    documents: List[Dict]
    total: int
    page: int
    page_size: int

class DocumentUpdateRequest(BaseModel):
    filename: Optional[str] = None
    metadata: Optional[Dict] = None
    status: Optional[str] = None

@router.get("", response_model=DocumentListResponse)
async def list_documents(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    status: Optional[str] = None,
    user_id: Optional[str] = None,
    token: str = Depends(oauth2_scheme)
):
    documents = []
    total = 0
    
    return DocumentListResponse(
        documents=documents,
        total=total,
        page=page,
        page_size=page_size
    )

@router.get("/{document_id}")
async def get_document(
    document_id: str,
    token: str = Depends(oauth2_scheme)
):
    return {}

@router.put("/{document_id}")
async def update_document(
    document_id: str,
    request: DocumentUpdateRequest,
    token: str = Depends(oauth2_scheme)
):
    return {}

@router.delete("/{document_id}")
async def delete_document(
    document_id: str,
    token: str = Depends(oauth2_scheme)
):
    return {}

@router.post("/{document_id}/reprocess")
async def reprocess_document(
    document_id: str,
    token: str = Depends(oauth2_scheme)
):
    return {}

用户管理API ​

python
from fastapi import APIRouter
from pydantic import BaseModel, Field

router = APIRouter(prefix="/api/admin/users", tags=["用户管理"])

class UserCreateRequest(BaseModel):
    username: str = Field(..., description="用户名")
    email: str = Field(..., description="邮箱")
    password: str = Field(..., description="密码")
    role: str = Field(..., description="角色")

class UserUpdateRequest(BaseModel):
    email: Optional[str] = None
    role: Optional[str] = None
    status: Optional[str] = None

@router.get("")
async def list_users(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    role: Optional[str] = None,
    status: Optional[str] = None,
    token: str = Depends(oauth2_scheme)
):
    return {}

@router.post("")
async def create_user(
    request: UserCreateRequest,
    token: str = Depends(oauth2_scheme)
):
    return {}

@router.get("/{user_id}")
async def get_user(
    user_id: str,
    token: str = Depends(oauth2_scheme)
):
    return {}

@router.put("/{user_id}")
async def update_user(
    user_id: str,
    request: UserUpdateRequest,
    token: str = Depends(oauth2_scheme)
):
    return {}

@router.delete("/{user_id}")
async def delete_user(
    user_id: str,
    token: str = Depends(oauth2_scheme)
):
    return {}

文档管理功能 ​

文档列表组件 ​

typescript
import React, { useState, useEffect } from 'react';
import { Table, Button, Space, Tag, Modal, message } from 'antd';
import { EyeOutlined, EditOutlined, DeleteOutlined, ReloadOutlined } from '@ant-design/icons';

interface Document {
  id: string;
  filename: string;
  file_size: number;
  upload_time: string;
  user_id: string;
  status: string;
  metadata: Record<string, any>;
}

interface DocumentListProps {
  apiBaseUrl?: string;
}

const DocumentList: React.FC<DocumentListProps> = ({
  apiBaseUrl = 'http://localhost:8000/api/admin/documents'
}) => {
  const [documents, setDocuments] = useState<Document[]>([]);
  const [loading, setLoading] = useState(false);
  const [pagination, setPagination] = useState({
    current: 1,
    pageSize: 20,
    total: 0
  });

  const fetchDocuments = async (page = 1, pageSize = 20) => {
    setLoading(true);
    try {
      const response = await fetch(
        `${apiBaseUrl}?page=${page}&page_size=${pageSize}`
      );
      const data = await response.json();

      setDocuments(data.documents);
      setPagination({
        current: page,
        pageSize,
        total: data.total
      });
    } catch (error) {
      message.error('获取文档列表失败');
    } finally {
      setLoading(false);
    }
  };

  useEffect(() => {
    fetchDocuments();
  }, []);

  const handleView = (document: Document) => {
    Modal.info({
      title: '文档详情',
      content: (
        <div>
          <p><strong>文件名：</strong>{document.filename}</p>
          <p><strong>文件大小：</strong>{(document.file_size / 1024).toFixed(2)} KB</p>
          <p><strong>上传时间：</strong>{document.upload_time}</p>
          <p><strong>状态：</strong>
            <Tag color={document.status === 'completed' ? 'green' : 'orange'}>
              {document.status}
            </Tag>
          </p>
        </div>
      ),
      width: 600
    });
  };

  const handleReprocess = async (documentId: string) => {
    try {
      const response = await fetch(
        `${apiBaseUrl}/${documentId}/reprocess`,
        { method: 'POST' }
      );
      
      if (response.ok) {
        message.success('文档重新处理成功');
        fetchDocuments(pagination.current, pagination.pageSize);
      } else {
        message.error('文档重新处理失败');
      }
    } catch (error) {
      message.error('文档重新处理失败');
    }
  };

  const handleDelete = async (documentId: string) => {
    Modal.confirm({
      title: '确认删除',
      content: '确定要删除这个文档吗？',
      onOk: async () => {
        try {
          const response = await fetch(
            `${apiBaseUrl}/${documentId}`,
            { method: 'DELETE' }
          );
          
          if (response.ok) {
            message.success('文档删除成功');
            fetchDocuments(pagination.current, pagination.pageSize);
          } else {
            message.error('文档删除失败');
          }
        } catch (error) {
          message.error('文档删除失败');
        }
      }
    });
  };

  const columns = [
    {
      title: '文件名',
      dataIndex: 'filename',
      key: 'filename'
    },
    {
      title: '文件大小',
      dataIndex: 'file_size',
      key: 'file_size',
      render: (size: number) => `${(size / 1024).toFixed(2)} KB`
    },
    {
      title: '上传时间',
      dataIndex: 'upload_time',
      key: 'upload_time',
      render: (time: string) => new Date(time).toLocaleString()
    },
    {
      title: '状态',
      dataIndex: 'status',
      key: 'status',
      render: (status: string) => (
        <Tag color={status === 'completed' ? 'green' : 'orange'}>
          {status}
        </Tag>
      )
    },
    {
      title: '操作',
      key: 'action',
      render: (_: any, record: Document) => (
        <Space size="middle">
          <Button
            type="link"
            icon={<EyeOutlined />}
            onClick={() => handleView(record)}
          >
            查看
          </Button>
          <Button
            type="link"
            icon={<ReloadOutlined />}
            onClick={() => handleReprocess(record.id)}
          >
            重新处理
          </Button>
          <Button
            type="link"
            danger
            icon={<DeleteOutlined />}
            onClick={() => handleDelete(record.id)}
          >
            删除
          </Button>
        </Space>
      )
    }
  ];

  return (
    <div>
      <Table
        columns={columns}
        dataSource={documents}
        rowKey="id"
        loading={loading}
        pagination={{
          current: pagination.current,
          pageSize: pagination.pageSize,
          total: pagination.total,
          onChange: (page, pageSize) => {
            fetchDocuments(page, pageSize);
          }
        }}
      />
    </div>
  );
};

export default DocumentList;

文档上传组件 ​

typescript
import React, { useState } from 'react';
import { Upload, Button, message, Progress } from 'antd';
import { UploadOutlined } from '@ant-design/icons';

interface DocumentUploadProps {
  apiBaseUrl?: string;
  onUploadSuccess?: () => void;
}

const DocumentUpload: React.FC<DocumentUploadProps> = ({
  apiBaseUrl = 'http://localhost:8000/api/documents',
  onUploadSuccess
}) => {
  const [uploading, setUploading] = useState(false);
  const [progress, setProgress] = useState(0);

  const handleUpload = async (file: File) => {
    const formData = new FormData();
    formData.append('file', file);

    setUploading(true);
    setProgress(0);

    try {
      const xhr = new XMLHttpRequest();

      xhr.upload.onprogress = (event) => {
        if (event.lengthComputable) {
          const percentComplete = (event.loaded / event.total) * 100;
          setProgress(percentComplete);
        }
      };

      xhr.onload = async () => {
        if (xhr.status === 200) {
          message.success('文档上传成功');
          setProgress(100);
          
          if (onUploadSuccess) {
            onUploadSuccess();
          }
        } else {
          message.error('文档上传失败');
        }
        setUploading(false);
      };

      xhr.onerror = () => {
        message.error('文档上传失败');
        setUploading(false);
      };

      xhr.open('POST', apiBaseUrl);
      xhr.send(formData);
    } catch (error) {
      message.error('文档上传失败');
      setUploading(false);
    }

    return false;
  };

  return (
    <div>
      <Upload
        beforeUpload={handleUpload}
        disabled={uploading}
        showUploadList={false}
      >
        <Button
          icon={<UploadOutlined />}
          loading={uploading}
        >
          {uploading ? '上传中...' : '上传文档'}
        </Button>
      </Upload>
      
      {uploading && (
        <Progress
          percent={progress}
          status="active"
          style={{ marginTop: 16 }}
        />
      )}
    </div>
  );
};

export default DocumentUpload;

用户管理功能 ​

用户列表组件 ​

typescript
import React, { useState, useEffect } from 'react';
import { Table, Button, Space, Tag, Modal, Form, Input, Select, message } from 'antd';
import { PlusOutlined, EditOutlined, DeleteOutlined } from '@ant-design/icons';

interface User {
  id: string;
  username: string;
  email: string;
  role: string;
  status: string;
  created_at: string;
}

interface UserListProps {
  apiBaseUrl?: string;
}

const UserList: React.FC<UserListProps> = ({
  apiBaseUrl = 'http://localhost:8000/api/admin/users'
}) => {
  const [users, setUsers] = useState<User[]>([]);
  const [loading, setLoading] = useState(false);
  const [modalVisible, setModalVisible] = useState(false);
  const [editingUser, setEditingUser] = useState<User | null>(null);
  const [form] = Form.useForm();

  const fetchUsers = async () => {
    setLoading(true);
    try {
      const response = await fetch(apiBaseUrl);
      const data = await response.json();

      setUsers(data.users || []);
    } catch (error) {
      message.error('获取用户列表失败');
    } finally {
      setLoading(false);
    }
  };

  useEffect(() => {
    fetchUsers();
  }, []);

  const handleCreate = () => {
    setEditingUser(null);
    form.resetFields();
    setModalVisible(true);
  };

  const handleEdit = (user: User) => {
    setEditingUser(user);
    form.setFieldsValue(user);
    setModalVisible(true);
  };

  const handleDelete = async (userId: string) => {
    Modal.confirm({
      title: '确认删除',
      content: '确定要删除这个用户吗？',
      onOk: async () => {
        try {
          const response = await fetch(
            `${apiBaseUrl}/${userId}`,
            { method: 'DELETE' }
          );
          
          if (response.ok) {
            message.success('用户删除成功');
            fetchUsers();
          } else {
            message.error('用户删除失败');
          }
        } catch (error) {
          message.error('用户删除失败');
        }
      }
    });
  };

  const handleSubmit = async () => {
    try {
      const values = await form.validateFields();

      if (editingUser) {
        const response = await fetch(
          `${apiBaseUrl}/${editingUser.id}`,
          {
            method: 'PUT',
            headers: { 'Content-Type': 'application/json' },
            body: JSON.stringify(values)
          }
        );

        if (response.ok) {
          message.success('用户更新成功');
          fetchUsers();
          setModalVisible(false);
        } else {
          message.error('用户更新失败');
        }
      } else {
        const response = await fetch(
          apiBaseUrl,
          {
            method: 'POST',
            headers: { 'Content-Type': 'application/json' },
            body: JSON.stringify(values)
          }
        );

        if (response.ok) {
          message.success('用户创建成功');
          fetchUsers();
          setModalVisible(false);
        } else {
          message.error('用户创建失败');
        }
      }
    } catch (error) {
      message.error('操作失败');
    }
  };

  const columns = [
    {
      title: '用户名',
      dataIndex: 'username',
      key: 'username'
    },
    {
      title: '邮箱',
      dataIndex: 'email',
      key: 'email'
    },
    {
      title: '角色',
      dataIndex: 'role',
      key: 'role',
      render: (role: string) => (
        <Tag color={role === 'admin' ? 'red' : 'blue'}>
          {role}
        </Tag>
      )
    },
    {
      title: '状态',
      dataIndex: 'status',
      key: 'status',
      render: (status: string) => (
        <Tag color={status === 'active' ? 'green' : 'default'}>
          {status}
        </Tag>
      )
    },
    {
      title: '创建时间',
      dataIndex: 'created_at',
      key: 'created_at',
      render: (time: string) => new Date(time).toLocaleString()
    },
    {
      title: '操作',
      key: 'action',
      render: (_: any, record: User) => (
        <Space size="middle">
          <Button
            type="link"
            icon={<EditOutlined />}
            onClick={() => handleEdit(record)}
          >
            编辑
          </Button>
          <Button
            type="link"
            danger
            icon={<DeleteOutlined />}
            onClick={() => handleDelete(record.id)}
          >
            删除
          </Button>
        </Space>
      )
    }
  ];

  return (
    <div>
      <div style={{ marginBottom: 16 }}>
        <Button
          type="primary"
          icon={<PlusOutlined />}
          onClick={handleCreate}
        >
          新建用户
        </Button>
      </div>

      <Table
        columns={columns}
        dataSource={users}
        rowKey="id"
        loading={loading}
      />

      <Modal
        title={editingUser ? '编辑用户' : '新建用户'}
        open={modalVisible}
        onOk={handleSubmit}
        onCancel={() => setModalVisible(false)}
        width={600}
      >
        <Form form={form} layout="vertical">
          <Form.Item
            name="username"
            label="用户名"
            rules={[{ required: true, message: '请输入用户名' }]}
          >
            <Input placeholder="请输入用户名" />
          </Form.Item>

          <Form.Item
            name="email"
            label="邮箱"
            rules={[
              { required: true, message: '请输入邮箱' },
              { type: 'email', message: '请输入有效的邮箱地址' }
            ]}
          >
            <Input placeholder="请输入邮箱" />
          </Form.Item>

          {!editingUser && (
            <Form.Item
              name="password"
              label="密码"
              rules={[{ required: true, message: '请输入密码' }]}
            >
              <Input.Password placeholder="请输入密码" />
            </Form.Item>
          )}

          <Form.Item
            name="role"
            label="角色"
            rules={[{ required: true, message: '请选择角色' }]}
          >
            <Select placeholder="请选择角色">
              <Select.Option value="admin">管理员</Select.Option>
              <Select.Option value="editor">编辑</Select.Option>
              <Select.Option value="viewer">查看者</Select.Option>
            </Select>
          </Form.Item>

          <Form.Item
            name="status"
            label="状态"
            rules={[{ required: true, message: '请选择状态' }]}
          >
            <Select placeholder="请选择状态">
              <Select.Option value="active">活跃</Select.Option>
              <Select.Option value="inactive">非活跃</Select.Option>
            </Select>
          </Form.Item>
        </Form>
      </Modal>
    </div>
  );
};

export default UserList;

部署配置 ​

Docker配置 ​

dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
yaml
version: '3.8'

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:password@postgres:5432/knowledge_base
      - REDIS_URL=redis://redis:6379
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    depends_on:
      - postgres
      - redis
      - chroma
    volumes:
      - ./data:/app/data
    restart: unless-stopped

  postgres:
    image: postgres:15
    environment:
      - POSTGRES_DB=knowledge_base
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=password
    volumes:
      - postgres_data:/var/lib/postgresql/data
    restart: unless-stopped

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data
    restart: unless-stopped

  chroma:
    image: chromadb/chroma:latest
    ports:
      - "8001:8000"
    volumes:
      - chroma_data:/chroma/chroma
    restart: unless-stopped

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
      - ./ssl:/etc/nginx/ssl
    depends_on:
      - api
    restart: unless-stopped

volumes:
  postgres_data:
  redis_data:
  chroma_data:

Kubernetes配置 ​

yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: knowledge-base
  namespace: knowledge-base
spec:
  replicas: 3
  selector:
    matchLabels:
      app: knowledge-base
  template:
    metadata:
      labels:
        app: knowledge-base
    spec:
      containers:
      - name: api
        image: your-registry/knowledge-base:latest
        ports:
        - containerPort: 8000
        env:
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: knowledge-base-secrets
              key: database-url
        - name: OPENAI_API_KEY
          valueFrom:
            secretKeyRef:
              name: knowledge-base-secrets
              key: openai-api-key
        resources:
          requests:
            memory: "512Mi"
            cpu: "500m"
          limits:
            memory: "1Gi"
            cpu: "1000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5

监控告警 ​

性能监控 ​

python
import time
from typing import Dict, List
from collections import deque

class PerformanceMonitor:
    def __init__(self, max_history: int = 1000):
        self.max_history = max_history
        self.metrics = {
            "query_times": deque(maxlen=max_history),
            "document_processing_times": deque(maxlen=max_history),
            "error_rates": deque(maxlen=max_history),
            "active_users": deque(maxlen=max_history)
        }
    
    def record_query_time(self, query_time: float):
        self.metrics["query_times"].append({
            "value": query_time,
            "timestamp": time.time()
        })
    
    def record_document_processing_time(self, processing_time: float):
        self.metrics["document_processing_times"].append({
            "value": processing_time,
            "timestamp": time.time()
        })
    
    def record_error(self, error_type: str):
        self.metrics["error_rates"].append({
            "error_type": error_type,
            "timestamp": time.time()
        })
    
    def record_active_users(self, count: int):
        self.metrics["active_users"].append({
            "value": count,
            "timestamp": time.time()
        })
    
    def get_statistics(self) -> Dict:
        stats = {}
        
        for metric_name, metric_data in self.metrics.items():
            if not metric_data:
                stats[metric_name] = {
                    "count": 0,
                    "avg": 0,
                    "min": 0,
                    "max": 0
                }
                continue
            
            values = [
                item["value"]
                for item in metric_data
                if "value" in item
            ]
            
            if values:
                stats[metric_name] = {
                    "count": len(values),
                    "avg": sum(values) / len(values),
                    "min": min(values),
                    "max": max(values)
                }
            else:
                stats[metric_name] = {
                    "count": len(metric_data),
                    "total": len(metric_data)
                }
        
        return stats

实践练习 ​

练习1：实现文档管理 ​

python
def implement_document_management():
    document_storage = DocumentStorage()
    vector_store = VectorStore()
    
    return document_storage, vector_store

练习2：实现用户管理 ​

python
def implement_user_management():
    user_manager = UserManager()
    
    return user_manager

练习3：部署知识库 ​

bash
docker-compose up -d

总结 ​

本节我们学习了企业知识库的管理后台与部署：

  1. 管理后台开发
  2. 文档管理功能
  3. 用户管理功能
  4. 部署配置
  5. 监控告警

管理后台和部署是知识库上线的关键环节。

参考资源 ​