写给工程师的 MacBook 商用级大模型知识库部署方案(上)

简介: 写给工程师的 MacBook 商用级大模型知识库部署方案(上)




本文介绍了如何在自己的 MacBook 上部署一套知识库方案辅助自己的知识管理工作,希望能给每位计划自己搭建大模型知识库应用的工程师一点参考。


背景


历史的车轮滚滚向前,大模型技术发展日新月异,每天都有新鲜的技术出炉,让人目不暇接,同时具备可玩性和想象空间的各种应用和开源库,仿佛让自己回到了第一次设置 JAVA_HOME 的日子,作为一枚古典工程师,我专门挑了个可能对手上工作有帮助的方向小试一把,尝试在自己的 MacBook 上部署一套知识库方案,看看能不能辅助自己的知识管理工作。


我自己的 Macbook 配置情况如下,可以流畅地运行没问题。经过量化处理的大模型,还是对办公本很友好的。



为什么要在 MacBook 搭建而不是直接采用现成的云服务呢?最核心最重要的是我们手上的文档资料出于安全要求,不能随便上传云服务,也就无法实际验证知识库的实际效用;另外对于工程师来说,自己亲手搭建一个完整的方案、能灵活调整和对接各种不同的模型、评测各种模型不同的表现,也是出于对技术的探索本能使然。



鉴于大模型已经是大模型及其周边概念已经是大家耳熟能详的东西,我这里就不再重复阐述相关的基础概念和理论了,直接进入动手环节,以用最快的速度部署起一个可用的知识库平台为目标,先用起来,再分各个环节优化。

方案概述

 应用架构


首先来看一下最终方案的应用架构是什么样子(下图)。在这套方案中,我们采用实力排上游、并且在使用上对学术和商业都友好的国产大模型 ChatGLM3-6B 对话模型和基于 m3e-base 模型的 embedding search RAG 方案;基于这两个模型封装和 ChatGPT 兼容的 API 接口协议;通过引入 One API 接口管理&分发系统,形成统一的 LLM 接口渠道管理平台规范,并把封装好的接口协议注册进去;搭建与 Dify.ai 齐名的开源大模型知识库平台管理系统 FastGPT,实现集私有知识数据源预处理、嵌入检索、大模型对话一体的完整知识库应用流程。麻雀虽小五脏俱全,最终形成一套既满足商用标准、又能在 MacBook 跑起来的的方案。虽然智能程度和实际需求还有一定差距,但至少我们在不用额外购买显卡或云服务的情况下,以最小成本部署运行、并且能导入实际业务数据(如语雀知识库)进行实操验证,值得每位工程师都来动手尝试一下。


 成型展示


在用户终端,我们基于 FastGPT 提供知识库管理及使用方案。引用其官网介绍:FastGPT 是一个基于 LLM 大语言模型的知识库问答系统,提供开箱即用的数据处理、模型调用等能力。同时可以通过 Flow 可视化进行工作流编排,从而实现复杂的问答场景。先放一张官网上的图片,来增加一点吸引朋友们动手操作的动力:

 部署要点


本套方案部署分为四个主要环节、14个具体步骤,只要一步步实操下去,每位朋友都可以在自己的本本上拥有属于自己的私有大模型知识库系统,步骤清单如下:

主要环节 详细步骤
一、准备大模型 1.1 下载对话语言模型 ChatGLM3-6B1.2 下载文本嵌入模型 m3e-base1.3 使用 chatglm.cpp 对 ChatGLM3-6B 进行量化加速1.4 验证模型问答效果
二、搭建模型API服务 2.1 搭建模型API2.2 搭建 One API 接口管理/分发系统2.3 验证模型接口能力
三、搭建知识库应用 3.1 安装 MongoDB3.2 安装 PostgreSQL & pgvector3.3 搭建 FastGPT 知识库问答系统3.4 验证模型对话能力
四、知识库问答实战 4.1 准备知识库语料4.2 导入知识库数据4.3 验证知识库问答效果


部分步骤可以简单地通过 Docker 镜像一键部署完成,但本着对细节一杆子插到底的部署思路,还是采取了纯手工作业的方法。注意,下面的步骤中仅包含了关键的命令,完整的命令可以参考对应系统的官网介绍。部分安装步骤如果速度不够理想,可以考虑采用国内源,包含但不限于 go、brew、pip、npm 等。


详细步骤


 准备离线模型


这个环节我们的主要任务是把模型文件准备好、完成量化,并通过命令行的方式,进行交互式对话验证。


  • 下载对话语言模型 ChatGLM3-6B


为什么选择 ChatGLM3-6B?常年霸榜的开源国产之光。ChatGLM3 一共开源了对话模型 ChatGLM-6B、基础模型 ChatGLM-6B-Base、长文本对话模型 ChatGLM3-6B-32K,对学术研究完全开放,在填写问卷进行登记后亦允许免费商业使用。无论是用来做上手实践还是微调练习,目前看来都是比较好的选择。
其实最重要的是,看看排行榜上的可选项,我的 MacBook 16G 内存只能带得动 ChatGLM3-6B 量化版本:

ChatGLM3-6B 现在比较方便的下载渠道有 HuggingFace 和 ModelScope,但是很明显能直接下载下来的可能性不大,所以我用家里的旧电脑科学下载后放到私有云CDN上,然后再用公司电脑下载,也方便未来随时随地取用,就是要花点小钱。ModelScope 也试过,不能直接下载文件,并且用 git clone 速度也不太理想,遂放弃。


如果用老一点的版本 ChatGLM2-6B 的话,网上也能找到一些比较好用的第三方镜像站。

  1. HuggingFace:THUDM/chatglm3-6b
  2. ModelScope:ZhipuAI/chatglm3-6b(地址:https://modelscope.cn/models/ZhipuAI/chatglm3-6b/summary


// 从 Git 仓库下载模型文件
// HuggingFace
git lfs install
git clone https://huggingface.co/THUDM/chatglm3-6b

// ModelScope
git lfs install
git clone https://www.modelscope.cn/ZhipuAI/chatglm3-6b.git


  • 下载文本嵌入模型 m3e-base


为什么选择 moka-ai 的 M3E 模型 m3e-base?M3E 向量模型属于小模型,资源使用不高,CPU 也可以运行,使用场景主要是中文,少量英文的情况。用来验证我们的知识库系统足够了

官方下载地址:moka-ai/m3e-base,先把所有的模型文件 download 下来,后面使用


  • 使用 chatglm.cpp 对 ChatGLM3-6B 进行量化加速


当我第一次知道 chatglm.cpp,只能说好人一生平安,chatglm.cpp 的出现拯救了纯 MacBook 党,让我们能在(低性能的)果本上基于 CPU 进行推理,也不会损失过多的精度。(其实损失多少我也不知道,不影响我们正常进行工程部署验证就行)Github Repo: https://github.com/li-plus/chatglm.cpp我使用的 Python 版本:3.11,最好单独准备一个 virtualenv

安装依赖:

cd /Users/yaolu/AGI/github/chatglm.cpp

# 先初始化 git 仓库
git submodule update --init --recursive

# 构建可执行文件
cmake -B build
cmake --build build -j

# 安装 Python 依赖
pip install .


如果发生 No module named 'chatglm_cpp._C' 的错误,把编译出来的文件 _C.cpython-311-darwin.so 放到 chatglm_cpp 目录下。


对 ChatGLM3-6B 进行 8-bit 量化处理:

python ./chatglm_cpp/convert.py -i /Users/yaolu/AGI/huggingface/THUDM/chatglm3-6b -t q8_0 -o chatglm3-ggml-q8.bin


如果电脑带不动,还可以尝试 4-bit、5-bit 参数量化,完整参数列表见 chatglm.cpp 的 quantization types

  • 验证模型问答效果


完成模型量化后,就可以在本地把大模型跑起来了,命令如下:

./build/bin/main -m chatglm3-ggml-q8.bin -i



 搭建模型API服务



我们在这个环节要完成的任务是,按照 ChatGPT 的接口规范、基于 FastAPI 封装 ChatGLM3-6B 的对话和 m3e-base 的嵌入能力;并注册到 One API 接口管理/分发系统中。

  • 搭建模型API


用 chatglm.cpp 自带的 openai_api.py 魔改了一下,使其支持完成对话和文本 embedding 的两个核心调用:

  1. /v1/chat/completions
  2. /v1/embeddings

代码如下:


import asyncio
import logging
import time
from typing import List, Literal, Optional, Union

import chatglm_cpp
from fastapi import FastAPI, HTTPException, status, Depends
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field#, computed_field
#from pydantic_settings import BaseSettings
from sse_starlette.sse import EventSourceResponse

from sentence_transformers import SentenceTransformer
from sklearn.preprocessing import PolynomialFeatures
import numpy as np
import tiktoken

logging.basicConfig(level=logging.INFO, format=r"%(asctime)s - %(module)s - %(levelname)s - %(message)s")


class Settings(object):
    model: str = "/Users/yaolu/AGI/github/chatglm.cpp/chatglm3-ggml-q8.bin"
    num_threads: int = 0


class ChatMessage(BaseModel):
    role: Literal["system", "user", "assistant"]
    content: str


class DeltaMessage(BaseModel):
    role: Optional[Literal["system", "user", "assistant"]] = None
    content: Optional[str] = None


class ChatCompletionRequest(BaseModel):
    model: str = "default-model"
    messages: List[ChatMessage]
    temperature: float = Field(default=0.95, ge=0.0, le=2.0)
    top_p: float = Field(default=0.7, ge=0.0, le=1.0)
    stream: bool = False
    max_tokens: int = Field(default=2048, ge=0)

    model_config = {
        "json_schema_extra": {"examples": [{"model": "default-model", "messages": [{"role": "user", "content": "你好"}]}]}
    }


class ChatCompletionResponseChoice(BaseModel):
    index: int = 0
    message: ChatMessage
    finish_reason: Literal["stop", "length"] = "stop"


class ChatCompletionResponseStreamChoice(BaseModel):
    index: int = 0
    delta: DeltaMessage
    finish_reason: Optional[Literal["stop", "length"]] = None


class ChatCompletionUsage(BaseModel):
    prompt_tokens: int
    completion_tokens: int

    #@computed_field
    @property
    def total_tokens(self) -> int:
        return self.prompt_tokens + self.completion_tokens


class ChatCompletionResponse(BaseModel):
    id: str = "chatcmpl"
    model: str = "default-model"
    object: Literal["chat.completion", "chat.completion.chunk"]
    created: int = Field(default_factory=lambda: int(time.time()))
    choices: Union[List[ChatCompletionResponseChoice], List[ChatCompletionResponseStreamChoice]]
    usage: Optional[ChatCompletionUsage] = None

    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "id": "chatcmpl",
                    "model": "default-model",
                    "object": "chat.completion",
                    "created": 1691166146,
                    "choices": [
                        {
                            "index": 0,
                            "message": {"role": "assistant", "content": "你好👋!我是人工智能助手 ChatGLM2-6B,很高兴见到你,欢迎问我任何问题。"},
                            "finish_reason": "stop",
                        }
                    ],
                    "usage": {"prompt_tokens": 17, "completion_tokens": 29, "total_tokens": 46},
                }
            ]
        }
    }


settings = Settings()
app = FastAPI()
app.add_middleware(
    CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"]
)
pipeline = chatglm_cpp.Pipeline(settings.model)
lock = asyncio.Lock()

embeddings_model = SentenceTransformer('/Users/yaolu/AGI/huggingface/moka-ai/m3e-base', device='cpu')


def stream_chat(history, body):
    yield ChatCompletionResponse(
        object="chat.completion.chunk",
        choices=[ChatCompletionResponseStreamChoice(delta=DeltaMessage(role="assistant"))],
    )

    for piece in pipeline.chat(
        history,
        max_length=body.max_tokens,
        do_sample=body.temperature > 0,
        top_p=body.top_p,
        temperature=body.temperature,
        num_threads=settings.num_threads,
        stream=True,
    ):
        yield ChatCompletionResponse(
            object="chat.completion.chunk",
            choices=[ChatCompletionResponseStreamChoice(delta=DeltaMessage(content=piece))],
        )

    yield ChatCompletionResponse(
        object="chat.completion.chunk",
        choices=[ChatCompletionResponseStreamChoice(delta=DeltaMessage(), finish_reason="stop")],
    )


async def stream_chat_event_publisher(history, body):
    output = ""
    try:
        async with lock:
            for chunk in stream_chat(history, body):
                await asyncio.sleep(0)  # yield control back to event loop for cancellation check
                output += chunk.choices[0].delta.content or ""
                yield chunk.model_dump_json(exclude_unset=True)
        logging.info(f'prompt: "{history[-1]}", stream response: "{output}"')
    except asyncio.CancelledError as e:
        logging.info(f'prompt: "{history[-1]}", stream response (partial): "{output}"')
        raise e


@app.post("/v1/chat/completions")
async def create_chat_completion(body: ChatCompletionRequest) -> ChatCompletionResponse:
    # ignore system messages
    history = [msg.content for msg in body.messages if msg.role != "system"]
    if len(history) % 2 != 1:
        raise HTTPException(status.HTTP_400_BAD_REQUEST, "invalid history size")

    if body.stream:
        generator = stream_chat_event_publisher(history, body)
        return EventSourceResponse(generator)

    max_context_length = 512
    output = pipeline.chat(
        history=history,
        max_length=body.max_tokens,
        max_context_length=max_context_length,
        do_sample=body.temperature > 0,
        top_p=body.top_p,
        temperature=body.temperature,
    )
    logging.info(f'prompt: "{history[-1]}", sync response: "{output}"')
    prompt_tokens = len(pipeline.tokenizer.encode_history(history, max_context_length))
    completion_tokens = len(pipeline.tokenizer.encode(output, body.max_tokens))

    return ChatCompletionResponse(
        object="chat.completion",
        choices=[ChatCompletionResponseChoice(message=ChatMessage(role="assistant", content=output))],
        usage=ChatCompletionUsage(prompt_tokens=prompt_tokens, completion_tokens=completion_tokens),
    )


class EmbeddingRequest(BaseModel):
    input: List[str]
    model: str

class EmbeddingResponse(BaseModel):
    data: list
    model: str
    object: str
    usage: dict

def num_tokens_from_string(string: str) -> int:
    """Returns the number of tokens in a text string."""
    encoding = tiktoken.get_encoding('cl100k_base')
    num_tokens = len(encoding.encode(string))
    return num_tokens

def expand_features(embedding, target_length):
    poly = PolynomialFeatures(degree=2)
    expanded_embedding = poly.fit_transform(embedding.reshape(1, -1))
    expanded_embedding = expanded_embedding.flatten()
    if len(expanded_embedding) > target_length:
        # 如果扩展后的特征超过目标长度,可以通过截断或其他方法来减少维度
        expanded_embedding = expanded_embedding[:target_length]
    elif len(expanded_embedding) < target_length:
        # 如果扩展后的特征少于目标长度,可以通过填充或其他方法来增加维度
        expanded_embedding = np.pad(expanded_embedding, (0, target_length - len(expanded_embedding)))
    return expanded_embedding


@app.post("/v1/embeddings", response_model=EmbeddingResponse)
async def get_embeddings(request: EmbeddingRequest):


    # 计算嵌入向量和tokens数量 
    embeddings = [embeddings_model.encode(text) for text in request.input]

    # 如果嵌入向量的维度不为1536,则使用插值法扩展至1536维度 
    embeddings = [expand_features(embedding, 1536) if len(embedding) < 1536 else embedding for embedding in embeddings]

    # Min-Max normalization
    embeddings = [embedding / np.linalg.norm(embedding) for embedding in embeddings]

    # 将numpy数组转换为列表
    embeddings = [embedding.tolist() for embedding in embeddings]
    prompt_tokens = sum(len(text.split()) for text in request.input)
    total_tokens = sum(num_tokens_from_string(text) for text in request.input)


    response = {
        "data": [
            {
                "embedding": embedding,
                "index": index,
                "object": "embedding"
            } for index, embedding in enumerate(embeddings)
        ],
        "model": request.model,
        "object": "list",
        "usage": {
            "prompt_tokens": prompt_tokens,
            "total_tokens": total_tokens,
        }
    }

    return response



class ModelCard(BaseModel):
    id: str
    object: Literal["model"] = "model"
    owned_by: str = "owner"
    permission: List = []


class ModelList(BaseModel):
    object: Literal["list"] = "list"
    data: List[ModelCard] = []

    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "object": "list",
                    "data": [{"id": "gpt-3.5-turbo", "object": "model", "owned_by": "owner", "permission": []}],
                }
            ]
        }
    }


@app.get("/v1/models")
async def list_models() -> ModelList:
    return ModelList(data=[ModelCard(id="gpt-3.5-turbo")])


让他跑起来的命令,跑在8000端口下:

uvicorn chatglm_cpp.openai_api:app --host 127.0.0.1 --port 8000


写给工程师的 MacBook 商用级大模型知识库部署方案(中):

https://developer.aliyun.com/article/1443297


目录
相关文章
|
11天前
|
机器学习/深度学习 数据采集 人工智能
文档智能 & RAG 让AI大模型更懂业务 —— 阿里云LLM知识库解决方案评测
随着数字化转型的深入,企业对文档管理和知识提取的需求日益增长。阿里云推出的文档智能 & RAG(Retrieval-Augmented Generation)解决方案,通过高效的内容清洗、向量化处理、精准的问答召回和灵活的Prompt设计,帮助企业构建强大的LLM知识库,显著提升企业级文档管理的效率和准确性。
|
14天前
|
机器学习/深度学习 数据采集 人工智能
大模型体验报告:阿里云文档智能 & RAG结合构建LLM知识库
大模型体验报告:阿里云文档智能 & RAG结合构建LLM知识库
|
5月前
|
人工智能 Linux Docker
一文详解几种常见本地大模型个人知识库工具部署、微调及对比选型(1)
近年来,大模型在AI领域崭露头角,成为技术创新的重要驱动力。从AlphaGo的胜利到GPT系列的推出,大模型展现出了强大的语言生成、理解和多任务处理能力,预示着智能化转型的新阶段。然而,要将大模型的潜力转化为实际生产力,需要克服理论到实践的鸿沟,实现从实验室到现实世界的落地应用。阿里云去年在云栖大会上发布了一系列基于通义大模型的创新应用,标志着大模型技术开始走向大规模商业化和产业化。这些应用展示了大模型在交通、电力、金融、政务、教育等多个行业的广阔应用前景,并揭示了构建具有行业特色的“行业大模型”这一趋势,大模型知识库概念随之诞生。
138262 30
|
4月前
|
Ubuntu API 数据安全/隐私保护
告别信息搜寻烦恼:用fastgpt快速部署国内大模型知识库助手
告别信息搜寻烦恼:用fastgpt快速部署国内大模型知识库助手
380 0
|
6月前
|
自然语言处理 API 开发工具
基于LangChain-Chatchat实现的本地知识库的问答应用-快速上手(检索增强生成(RAG)大模型)
基于LangChain-Chatchat实现的本地知识库的问答应用-快速上手(检索增强生成(RAG)大模型)
基于LangChain-Chatchat实现的本地知识库的问答应用-快速上手(检索增强生成(RAG)大模型)
|
6月前
|
存储 人工智能 搜索推荐
社区供稿 | YuanChat全面升级:知识库、网络检索、适配CPU,手把手个人主机部署使用教程
在当下大语言模型飞速发展的背景下,以大模型为核心的AI助手成为了广大企业和个人用户最急切需求的AI产品。然而在复杂的现实办公场景下,简单的对话功能并不能满足用户的全部办公需求,为此我们发布了最新版的YuanChat应用
|
5月前
|
人工智能 小程序 机器人
开源一个RAG大模型本地知识库问答机器人-ChatWiki
准备工作 再安装ChatWiki之前,您需要准备一台具有联网功能的linux服务器,并确保服务器满足最低系统要求 • Cpu:最低需要2 Core • RAM:最低需要4GB 开始安装 ChatWiki社区版基于Docker部署,请先确保服务器已经安装好Docker。如果没有安装,可以通过以下命令安装:
292 0
|
6月前
|
机器学习/深度学习 自然语言处理 机器人
实时数仓 Hologres产品使用合集之业级问答知识库该如何部署有教程吗
实时数仓Hologres是阿里云推出的一款高性能、实时分析的数据库服务,专为大数据分析和复杂查询场景设计。使用Hologres,企业能够打破传统数据仓库的延迟瓶颈,实现数据到决策的无缝衔接,加速业务创新和响应速度。以下是Hologres产品的一些典型使用场景合集。
132 0
|
2月前
|
人工智能 JSON 数据格式
RAG+Agent人工智能平台:RAGflow实现GraphRA知识库问答,打造极致多模态问答与AI编排流体验
【9月更文挑战第6天】RAG+Agent人工智能平台:RAGflow实现GraphRA知识库问答,打造极致多模态问答与AI编排流体验
RAG+Agent人工智能平台:RAGflow实现GraphRA知识库问答,打造极致多模态问答与AI编排流体验
|
3月前
|
开发框架 自然语言处理 API
基于RAG搭建企业级知识库在线问答
本文介绍如何使用搜索开发工作台快速搭建基于RAG开发链路的知识库问答应用。
8268 17

热门文章

最新文章