本文实现了基于langchain 的本地知识库的基本功能，可离线访问。主要使用了Langchain，ChromaDb Ollama。

本文实现了基于langchain 的本地知识库的基本功能，可离线访问。主要使用了Langchain，ChromaDb Ollama。

概念介绍

什么是RAG

RAG是retrieval-augmented-generation的缩写，直译中文的意思是 检索增强生成，可以简单理解能让训练好的大模型 LLM可以结合外部数据，可以补充或者修正大模型返回的答案，提高答案的准确性。

LLM有哪些痛点

开始实现实现步骤

我们目前要做的是

前期准备

Ollama

安装 ollama 官网上有，这里就多说了。这里使用到了两个模型，可以用ollama pull 拉下拉

ChromaDb

它支持本地存储以及独立客户端，选择本地存储的化不需要安装客户端参考 /deployment

Hugging Face

用 huggingface-cli 下载 nomic-embed-text 作为 tokenizer 的模型

开始开发数据准备和存储

文档读取

from langchain_community.document_loaders import TextLoader
loader = TextLoader(file_path=self.file_path, autodetect_encoding=True)
documents = loader.load()

文档拆分这里使用的是本地模型。实现了两个方法，拆分字符串和拆分文档

from typing import Any, Iterable, List
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
from langchain.text_splitter import CharacterTextSplitter
from transformers import AutoTokenizer
huggging_model_path = "model_path"
class HuggingFaceTextSplitter(CharacterTextSplitter):
  
    def __init__(self,
                 separator: str = "nn",
                 is_separator_regex: bool = False,
                 **kwargs: Any) -> None:
        super().__init__(separator, is_separator_regex, **kwargs)
        
        self.tokenizer = AutoTokenizer.from_pretrained(huggging_model_path)
        self.text_splitter = RecursiveCharacterTextSplitter.from_huggingface_tokenizer(
            self.tokenizer, chunk_size=1000, chunk_overlap=100)
    def split_text(self, text: str) -> List[str]:
        return self.text_splitter.split_text(text) 
    def split_documents(self, documents: Iterable[Document]) -> List[Document]:
        return self.text_splitter.split_documents(documents)
        

huggging_model_path 这是一个本地模型的文件夹地址

文档向量化，存入数据库这里向量化的过程，没有单独实现，用的是数据库的特性，插入数据时自动实现我们先写一段简单的数据库service，做数据库的插入

from typing import List
import chromadb
from chromadb.api import ClientAPI
from chromadb.api.types import GetResult
from langchain.docstore.document import Document
from lib import ConfigService, get_Embeddings_func
from typing import Dict, List
import uuid
class ChromaService():
    data_path_base: str
    client: ClientAPI
    collection_name: str
    def __init__(self, collection_name: str) -> None:
    
        configService = ConfigService()
        self.data_path_base = configService.env_config.chromadb.get(
            "data_path_base")
        # 持久化存储
        self.client = chromadb.PersistentClient(path=self.data_path_base)
        self.embeddings_func = get_Embeddings_func()
        self.collection_name = collection_name
        self.collection = self.client.get_or_create_collection(
            collection_name, embedding_function=self.embeddings_func)
    def get_result_by_ids(self, ids: List[str]) -> GetResult:
    
        return self.collection.get(ids, include=["embeddings"])
    def add_doc(self, docs: List[Document], **kwargs) -> List[Dict]:
        doc_infos = []

        texts = [doc.page_content for doc in docs]
        metadatas = [doc.metadata for doc in docs]
        ids = [str(uuid.uuid1()) for _ in range(len(texts))]
        
        for _id, text, metadata in zip(ids, texts, metadatas):
            self.collection.add(
                ids=_id,
                metadatas=metadata,
                documents=text)
            doc_infos.append({"id": _id, "metadata": metadata})
        return doc_infos

这里数据库配置用了本地存储 PersistentClient，获取集合的时候指定了 embedding_function，这的用处是在数据插入的时候由数据库生成向量化数据，如果这个参数不传的化，就要手动插入向量数据。

from chromadb.utils.embedding_functions.ollama_embedding_function import OllamaEmbeddingFunction
def get_Embeddings_func():
    api_url="http://localhost:11434/api/embeddings"
    ollama_ef = OllamaEmbeddingFunction(
        url=api_url,
        model_name="nomic-embed-text",
    )
    return ollama_ef

数据至此已存入数据库

数据查询

在 ChromaService 里增加几个查询方法，首先是最简单的按照Id 查询

    def get_result_by_ids(self, ids: List[str]) -> GetResult:
    
        return self.collection.get(ids=ids)

其次是向量查询

    async def do_search_db(self, query) -> QueryResult::
    
        embed_query = await self.embeddings.aembed_query(query)
        result = self.collection.query(query_embeddings=embed_query)
        return result

这里的embeddings 来源于这里

from langchain_core.embeddings import Embeddings
from langchain_ollama import OllamaEmbeddings
def get_Embeddings() -> Embeddings:
    embeddings = OllamaEmbeddings(
        model="nomic-embed-text")
    return embeddings

和 LLM 结合

增加QaChain

在ChromaService 增加方法并且在 __init__ 里调用。

	from langchain_ollama import ChatOllama, OllamaEmbeddings
	from langchain_chroma import Chroma
	from langchain.chains.qa_with_sources.retrieval import RetrievalQAWithSourcesChain
	
    def _init_qa(self):
        vector_em = OllamaEmbeddings(model='nomic-embed-text')
        
        vectordb = Chroma(persist_directory=self.data_path_base,
                          collection_name=self.collection_name,
                          embedding_function=vector_em)
        ollamaLlm = ChatOllama(model="qwen2.5:3b",
                               custom_get_token_ids=get_token_encode)
        
        self.qa_retrieval = RetrievalQAWithSourcesChain.from_llm(
            llm=ollamaLlm,
            retriever=vectordb.as_retriever(search_type="similarity",
                                            search_kwargs={"k": 2}),
            return_source_documents=True)

这里有几点要说下

def get_token_encode(text: str) -> list[int]:
	huggging_model_path = "model_path"
    tokenizer = AutoTokenizer.from_pretrained(huggging_model_path)
    result = tokenizer.encode(text)
    return result

现在提问就很简单了query 就是你问题，一个字符串

  result = self.qa_retrieval.invoke(query)

result 就是给你的答案