MaxKB API全解析：开发者集成指南

2026-02-04 04:11:51作者：秋阔奎Evelyn

概述

MaxKB（Max Knowledge Brain）是一个强大易用的企业级智能体平台，基于LLM大语言模型的知识库问答系统。本文将为开发者提供完整的API集成指南，帮助您快速将MaxKB集成到第三方业务系统中。

API架构概览

MaxKB采用RESTful API设计，基于Django REST Framework构建，提供完整的API文档和类型安全支持。系统主要包含以下API模块：

graph TB
    A[MaxKB API架构] --> B[认证模块]
    A --> C[知识库管理]
    A --> D[文档管理]
    A --> E[段落管理]
    A --> F[问题管理]
    A --> G[聊天对话]
    A --> H[模型管理]
    A --> I[系统管理]
    
    B --> B1[用户认证]
    B --> B2[权限控制]
    
    C --> C1[知识库创建]
    C --> C2[知识库查询]
    C --> C3[知识库同步]
    
    G --> G1[对话创建]
    G --> G2[历史记录]
    G --> G3[实时流式响应]

核心API端点

1. 认证API

用户登录认证

# 请求示例
import requests

def login(username, password):
    url = "http://your-maxkb-domain/admin/api/login/"
    payload = {
        "username": username,
        "password": password
    }
    response = requests.post(url, json=payload)
    return response.json()

# 响应格式
{
    "code": 200,
    "message": "success",
    "data": {
        "token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
        "user": {
            "id": "user-uuid",
            "username": "admin",
            "email": "admin@example.com"
        }
    }
}

Token认证头

Authorization: Bearer {access_token}

2. 知识库管理API

创建知识库

POST /admin/api/knowledge/{workspace_id}/
Content-Type: application/json

{
    "name": "产品文档知识库",
    "desc": "包含所有产品文档和FAQ",
    "embedding_model_id": "model-uuid",
    "llm_model_id": "llm-model-uuid"
}

查询知识库列表

GET /admin/api/knowledge/{workspace_id}/

知识库分页查询

GET /admin/api/knowledge/{workspace_id}/{current_page}/{page_size}?folder_id={folder_id}

3. 文档管理API

上传文档

# 多部分表单数据上传
def upload_document(workspace_id, knowledge_id, file_path):
    url = f"http://your-maxkb-domain/admin/api/knowledge/{workspace_id}/{knowledge_id}/document/"
    files = {'file': open(file_path, 'rb')}
    data = {
        'name': '产品手册.pdf',
        'source_type': 'file'
    }
    response = requests.post(url, files=files, data=data)
    return response.json()

支持的文档类型

格式类型	支持情况	说明
PDF	✅	支持文本提取和目录解析
Word (.docx)	✅	完整格式支持
Excel (.xlsx)	✅	表格数据解析
CSV	✅	结构化数据处理
HTML	✅	网页内容抓取
Text	✅	纯文本文件
Markdown	✅	Markdown格式解析

4. 段落管理API

创建段落

POST /admin/api/knowledge/{workspace_id}/{knowledge_id}/{document_id}/paragraph/
Content-Type: application/json

{
    "title": "安装指南",
    "content": "1. 下载安装包...",
    "position": 1
}

批量操作段落

PUT /admin/api/knowledge/{workspace_id}/{knowledge_id}/{document_id}/paragraph/batch
Content-Type: application/json

{
    "operation": "generate_related",
    "paragraph_ids": ["para-uuid1", "para-uuid2"]
}

5. 聊天对话API

创建对话

POST /chat/api/chat/{chat_id}/
Content-Type: application/json

{
    "message": "如何安装MaxKB？",
    "application_id": "app-uuid",
    "stream": true
}

流式响应处理

def stream_chat(chat_id, message):
    url = f"http://your-maxkb-domain/chat/api/chat/{chat_id}/"
    payload = {
        "message": message,
        "stream": True
    }
    
    with requests.post(url, json=payload, stream=True) as response:
        for chunk in response.iter_lines():
            if chunk:
                data = json.loads(chunk.decode('utf-8'))
                yield data

6. 向量搜索API

语义搜索

POST /admin/api/knowledge/{workspace_id}/{knowledge_id}/hit_test/
Content-Type: application/json

{
    "query_text": "安装部署",
    "top_number": 5,
    "similarity": 0.7
}

API响应格式规范

所有API响应都遵循统一的格式：

{
    "code": 200,
    "message": "success",
    "data": {
        // 具体业务数据
    }
}

状态码说明

状态码	说明	处理建议
200	成功	正常处理返回数据
400	请求参数错误	检查请求参数格式
401	未授权	检查Token是否有效
403	权限不足	检查用户权限设置
404	资源不存在	检查资源ID是否正确
500	服务器内部错误	联系系统管理员

高级集成示例

1. 完整的知识库集成流程

class MaxKBClient:
    def __init__(self, base_url, token):
        self.base_url = base_url
        self.headers = {
            'Authorization': f'Bearer {token}',
            'Content-Type': 'application/json'
        }
    
    def create_knowledge_base(self, workspace_id, name, description):
        """创建知识库"""
        url = f"{self.base_url}/admin/api/knowledge/{workspace_id}/"
        payload = {
            "name": name,
            "desc": description,
            "embedding_model_id": self.get_default_embedding_model()
        }
        response = requests.post(url, json=payload, headers=self.headers)
        return response.json()
    
    def upload_documents(self, workspace_id, knowledge_id, file_paths):
        """批量上传文档"""
        results = []
        for file_path in file_paths:
            url = f"{self.base_url}/admin/api/knowledge/{workspace_id}/{knowledge_id}/document/"
            files = {'file': open(file_path, 'rb')}
            data = {'name': os.path.basename(file_path)}
            response = requests.post(url, files=files, data=data, headers=self.headers)
            results.append(response.json())
        return results
    
    def query_knowledge(self, workspace_id, knowledge_id, question):
        """查询知识库"""
        url = f"{self.base_url}/admin/api/knowledge/{workspace_id}/{knowledge_id}/hit_test/"
        payload = {
            "query_text": question,
            "top_number": 3,
            "similarity": 0.6
        }
        response = requests.post(url, json=payload, headers=self.headers)
        return response.json()

2. 实时聊天集成

// 前端集成示例
class MaxKBChat {
    constructor(apiUrl, applicationId) {
        this.apiUrl = apiUrl;
        this.applicationId = applicationId;
        this.chatId = this.generateChatId();
    }
    
    async sendMessage(message, onChunk, onComplete) {
        const response = await fetch(`${this.apiUrl}/chat/api/chat/${this.chatId}/`, {
            method: 'POST',
            headers: {
                'Content-Type': 'application/json',
                'Authorization': `Bearer ${localStorage.getItem('token')}`
            },
            body: JSON.stringify({
                message: message,
                application_id: this.applicationId,
                stream: true
            })
        });
        
        const reader = response.body.getReader();
        const decoder = new TextDecoder();
        let fullResponse = '';
        
        while (true) {
            const { value, done } = await reader.read();
            if (done) break;
            
            const chunk = decoder.decode(value);
            const lines = chunk.split('\n');
            
            for (const line of lines) {
                if (line.startsWith('data: ')) {
                    try {
                        const data = JSON.parse(line.slice(6));
                        if (data.content) {
                            fullResponse += data.content;
                            onChunk(data.content);
                        }
                        if (data.is_end) {
                            onComplete(fullResponse);
                        }
                    } catch (e) {
                        console.error('Parse error:', e);
                    }
                }
            }
        }
    }
}

错误处理最佳实践

1. 重试机制

def api_call_with_retry(url, payload, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = requests.post(url, json=payload, timeout=30)
            if response.status_code == 200:
                return response.json()
            elif response.status_code == 401:
                # Token过期，需要重新认证
                refresh_token()
                continue
        except requests.exceptions.RequestException as e:
            if attempt == max_retries - 1:
                raise e
            time.sleep(2 ** attempt)  # 指数退避

2. 限流处理

from ratelimit import limits, sleep_and_retry

@sleep_and_retry
@limits(calls=100, period=60)  # 每分钟100次调用
def call_maxkb_api(url, payload):
    response = requests.post(url, json=payload)
    return response.json()

性能优化建议

1. 批量操作

尽量使用批量API减少请求次数：

批量上传文档
批量处理段落
批量生成相关问题

2. 异步处理

对于耗时操作，使用异步任务：

# 使用Celery进行异步处理
@app.task
def async_embedding_task(document_ids):
    for doc_id in document_ids:
        # 调用嵌入API
        pass

3. 缓存策略

from django.core.cache import cache

def get_cached_knowledge(knowledge_id):
    cache_key = f"knowledge_{knowledge_id}"
    cached_data = cache.get(cache_key)
    if cached_data:
        return cached_data
    
    # 从API获取数据
    data = get_knowledge_from_api(knowledge_id)
    cache.set(cache_key, data, timeout=300)  # 缓存5分钟
    return data

安全注意事项

1. Token管理

定期轮换访问令牌
使用HTTPS加密传输
避免在前端存储敏感令牌

2. 输入验证

def validate_api_input(data, required_fields):
    for field in required_fields:
        if field not in data:
            raise ValueError(f"Missing required field: {field}")
        if not isinstance(data[field], str) or not data[field].strip():
            raise ValueError(f"Invalid value for field: {field}")

3. 权限控制

确保每个API调用都进行适当的权限检查：

def check_permission(user_id, resource_type, resource_id, action):
    # 调用权限验证API
    pass

监控和日志

1. API调用监控

import logging
import time

def api_call_with_monitoring(url, payload):
    start_time = time.time()
    try:
        response = requests.post(url, json=payload)
        duration = time.time() - start_time
        logging.info(f"API call to {url} completed in {duration:.2f}s")
        return response
    except Exception as e:
        logging.error(f"API call to {url} failed: {str(e)}")
        raise

2. 性能指标收集

from prometheus_client import Counter, Histogram

API_CALLS = Counter('maxkb_api_calls_total', 'Total API calls', ['endpoint', 'status'])
API_DURATION = Histogram('maxkb_api_duration_seconds', 'API call duration', ['endpoint'])

@API_DURATION.time()
def monitored_api_call(url, payload):
    response = requests.post(url, json=payload)
    API_CALLS.labels(endpoint=url, status=response.status_code).inc()
    return response