别再花几十万买商业知识库了！1Panel开源的这个项目，让我零成本搭建了一套生产级RAG系统

# 安装Docker（以Ubuntu为例）
curl -fsSL https://get.docker.com | sh

# 启动Docker服务
sudo systemctl start docker
sudo systemctl enable docker

# 安装Docker Compose
sudo curl -L "https://github.com/docker/compose/releases/download/v2.20.0/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose
sudo chmod +x /usr/local/bin/docker-compose

# 克隆仓库
git clone https://github.com/1Panel-dev/MaxKB.git

# 进入项目目录
cd MaxKB

# 如果下载的是单独的docker-compose文件，可以直接使用
# wget https://raw.githubusercontent.com/1Panel-dev/MaxKB/main/docker/docker-compose.yml

# 使用Docker Compose启动所有服务
docker-compose up -d

# 查看运行状态
docker-compose ps

http://你的服务器IP:8080

# 拉取所需镜像
docker pull maxkb/maxkb:latest
docker pull postgres:15-alpine
docker pull redis:7-alpine

# 导出镜像为tar文件
docker save -o maxkb-images.tar maxkb/maxkb:latest postgres:15-alpine redis:7-alpine

# 使用scp或其他方式传输
scp maxkb-images.tar user@目标服务器:/path/to/destination/

# 加载镜像
docker load -i maxkb-images.tar

# 验证镜像是否加载成功
docker images

# 检查容器状态
docker ps

# 查看日志
docker logs maxkb-postgres -f  # 检查数据库
docker logs maxkb-redis -f     # 检查缓存
docker logs maxkb-server -f    # 检查后端服务
docker logs maxkb-web -f       # 检查前端服务

你是一个专业的技术支持助手。你的职责是根据提供的知识库内容，准确回答用户的技术问题。

回答规范：
1. 只使用知识库中提供的信息进行回答
2. 如果知识库中没有相关信息，明确告知用户
3. 回答要清晰、简洁，包含具体的操作步骤
4. 如果问题涉及多个知识库内容，进行整合回答

import requests

# API配置
base_url = "http://你的MaxKB服务器地址"
client_id = "你的ClientID"
client_secret = "你的ClientSecret"

# 获取访问令牌
def get_access_token():
    url = f"{base_url}/api/v1/auth/verification"
    headers = {
        "Content-Type": "application/json"
    }
    data = {
        "client_id": client_id,
        "client_secret": client_secret
    }

    response = requests.post(url, json=data, headers=headers)
    if response.status_code == 200:
        return response.json().get("access_token")
    else:
        raise Exception(f"获取令牌失败: {response.text}")

access_token = get_access_token()
print(f"获取到的访问令牌: {access_token}")

import requests

# 创建新的对话会话
def create_chat_session(access_token, application_id):
    url = f"{base_url}/api/v1/application/{application_id}/chat/create"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {access_token}"
    }

    response = requests.post(url, headers=headers)
    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"创建会话失败: {response.text}")

# 使用示例
session_info = create_chat_session(access_token, "你的应用ID")
session_id = session_info.get("session_id")
print(f"会话ID: {session_id}")

import requests
import json

def send_message(access_token, application_id, session_id, message):
    url = f"{base_url}/api/v1/application/{application_id}/chat/message"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {access_token}"
    }
    data = {
        "session_id": session_id,
        "message": message,
        "stream": False  # 设置为True可启用流式输出
    }

    response = requests.post(url, json=data, headers=headers)
    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"发送消息失败: {response.text}")

# 发送一个技术问题
question = "产品支持哪些操作系统？"
answer = send_message(access_token, "你的应用ID", session_id, question)

print("问题:", question)
print("回答:", answer.get("answer"))
print("引用知识库:", answer.get("reference", []))

import requests
import json

def stream_chat(access_token, application_id, session_id, message):
    url = f"{base_url}/api/v1/application/{application_id}/chat/message"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {access_token}"
    }
    data = {
        "session_id": session_id,
        "message": message,
        "stream": True
    }

    response = requests.post(url, json=data, headers=headers, stream=True)

    if response.status_code == 200:
        for line in response.iter_lines():
            if line:
                # 解析SSE格式的数据
                line_str = line.decode('utf-8')
                if line_str.startswith('data:'):
                    json_str = line_str[5:].strip()
                    if json_str:
                        chunk = json.loads(json_str)
                        yield chunk
    else:
        raise Exception(f"流式请求失败: {response.text}")

# 使用示例：打印流式回答
full_answer = ""
for chunk in stream_chat(access_token, "你的应用ID", session_id, "如何联系技术支持？"):
    if chunk.get('type') == 'answer':
        content = chunk.get('content', '')
        print(content, end='', flush=True)
        full_answer += content
    elif chunk.get('type') == 'end':
        print("\n回答完成")

from flask import Flask, render_template, request, jsonify
import requests

app = Flask(__name__)

# 配置
MAXKB_URL = "http://你的MaxKB服务器地址"
CLIENT_ID = "你的ClientID"
CLIENT_SECRET = "你的ClientSecret"

# 获取访问令牌
def get_access_token():
    url = f"{MAXKB_URL}/api/v1/auth/verification"
    data = {"client_id": CLIENT_ID, "client_secret": CLIENT_SECRET}
    response = requests.post(url, json=data)
    return response.json().get("access_token")

# 初始化访问令牌
access_token = get_access_token()

@app.route('/')
def index():
    return render_template('chat.html')

@app.route('/api/chat', methods=['POST'])
def chat():
    user_message = request.json.get('message')
    session_id = request.json.get('session_id')

    # 创建新会话或使用已有会话
    if not session_id:
        create_url = f"{MAXKB_URL}/api/v1/application/你的应用ID/chat/create"
        headers = {"Authorization": f"Bearer {access_token}"}
        session_response = requests.post(create_url, headers=headers)
        session_id = session_response.json().get("session_id")

    # 发送消息
    chat_url = f"{MAXKB_URL}/api/v1/application/你的应用ID/chat/message"
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {access_token}"
    }
    data = {
        "session_id": session_id,
        "message": user_message
    }

    response = requests.post(chat_url, json=data, headers=headers)
    result = response.json()

    return jsonify({
        "answer": result.get("answer"),
        "session_id": session_id,
        "reference": result.get("reference", [])
    })

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

你是一个专业的技术支持工程师。请根据提供的知识库内容，准确回答用户的技术问题。

回答要求：
1. 回答必须基于知识库内容，不要编造信息
2. 涉及操作步骤时，使用清晰的编号列表
3. 对于不确定的问题，诚实告知用户无法从知识库中找到相关信息
4. 保持专业、友好的语气
5. 如有必要，提示用户查看相关文档链接

你是一个热情的产品顾问。你的目标是帮助用户了解产品特点，找到最适合他们需求的解决方案。

沟通原则：
1. 用通俗易懂的语言解释专业术语
2. 主动了解用户的需求和场景
3. 给出具体、可操作的建议
4. 如涉及多个产品选项，进行客观对比
5. 结束时询问用户是否还有其他问题

你是一个企业内部知识助手。请根据公司知识库为员工提供准确的信息支持。

行为规范：
1. 回答简洁明了，直奔主题
2. 如涉及流程或政策，明确说明适用范围
3. 对于敏感信息，仅提供公开范围的内容
4. 如需跨部门协调，提示员工联系相关人员
5. 鼓励员工提出改进建议

你同时拥有关于产品技术、公司政策和员工培训的完整知识库。
当用户提问时，请首先判断问题属于哪个类别：
- 技术问题：查阅产品技术文档库
- 政策问题：查阅公司政策库
- 培训相关：查阅培训资料库
- 综合性问题：综合多个知识库回答

-- 定期执行的维护任务
-- 清理 dead tuples
VACUUM FULL;

-- 更新统计信息
ANALYZE;

-- 检查索引使用情况
SELECT indexrelname, idx_scan, idx_tup_read 
FROM pg_stat_user_indexes 
ORDER BY idx_scan ASC;

# 示例：配置Redis缓存
import redis

redis_client = redis.Redis(
    host='你的Redis地址',
    port=6379,
    db=0,
    password='你的密码'
)

# 设置缓存，1小时过期
def cache_answer(question, answer):
    cache_key = f"qa:{hash(question)}"
    redis_client.setex(cache_key, 3600, answer)

# 获取缓存
def get_cached_answer(question):
    cache_key = f"qa:{hash(question)}"
    return redis_client.get(cache_key)

# 数据库备份脚本示例
#!/bin/bash
# 备份时间戳
BACKUP_DATE=$(date +%Y%m%d_%H%M%S)
BACKUP_DIR=/data/backup/maxkb

# 创建备份目录
mkdir -p $BACKUP_DIR

# 备份PostgreSQL数据
docker exec maxkb-postgres pg_dump -U postgres maxkb > $BACKUP_DIR/postgres_$BACKUP_DATE.sql

# 备份Redis数据
docker exec maxkb-redis redis-cli SAVE
docker cp maxkb-redis:/data/dump.rdb $BACKUP_DIR/redis_$BACKUP_DATE.rdb

# 清理超过30天的备份
find $BACKUP_DIR -type f -mtime +30 -delete

echo "备份完成: $BACKUP_DATE"