前言
過去半年，隨著ChatGPT的火爆，直接帶火了整個LLM這個方向，然LLM畢竟更多是基于過去的經驗數據預訓練而來，沒法獲取最新的知識，以及各企業(yè)私有的知識
- 為了獲取最新的知識，ChatGPT plus版集成了bing搜索的功能，有的模型則會調用一個定位于 “鏈接各種AI模型、工具”的langchain的bing功能
- 為了處理企業(yè)私有的知識，要么基于開源模型微調，要么更可以基于langchain里集成的向量數據庫和LLM搭建本地知識庫問答(此處的向量數據庫的獨特性在哪呢？舉個例子，傳統(tǒng)數據庫做圖片檢索可能是通過關鍵詞去搜索，向量數據庫是通過語義搜索圖片中相同或相近的向量并呈現結果)
所以越來越多的人開始關注langchain并把它與LLM結合起來應用，更直接推動了數據庫、知識圖譜與LLM的結合應用(詳見下一篇文章：知識圖譜實戰(zhàn)導論：從什么是KG到LLM與KG/DB的結合實戰(zhàn))
本文則側重講解
- 什么是LangChain及l(fā)angchain的整體組成架構
- 通過langchain-ChatGLM構建本地知識庫問答的基本流程，與每個流程背后的邏輯
- 解讀langchain-ChatGLM項目的關鍵源碼，不只是把它當做一個工具使用，因為對工具的原理更了解，則對工具的使用更順暢
一開始解讀不易，因為涉及的項目、技術點不少，所以一開始容易繞暈，好在根據該項目的流程一步步抽絲剝繭之后，給大家呈現了清晰的代碼架構
過程中，我從接觸該langchain-ChatGLM項目到整體源碼梳理清晰并寫清楚歷時了近一周，而大家有了本文之后，可能不到一天便可以理清了(提升近7倍效率) ???，這便是本文的價值和意義之一 - langchain-ChatGLM項目的升級版：langchain-Chatchat
- 我司基于langchain-chatchat二次開發(fā)的企業(yè)多文檔知識庫問答系統(tǒng)
閱讀過程中若有任何問題，歡迎隨時留言，會一一及時回復/解答，共同探討、共同深挖
第一部分?LangChain的整體組成架構：LLM的外掛/功能庫?
通俗講，所謂langchain (官網地址、GitHub地址)，即把AI中常用的很多功能都封裝成庫，且有調用各種商用模型API、開源模型的接口，支持以下各種組件
?初次接觸的朋友一看這么多組件可能直接暈了(封裝的東西非常多，感覺它想把LLM所需要用到的功能/工具都封裝起來)，為方便理解，我們可以先從大的層面把整個langchain庫劃分為三個大層：基礎層、能力層、應用層
1.1 基礎層：models、LLMs、index
1.1.1?Models：模型
各種類型的模型和模型集成，比如OpenAI的各個API/GPT-4等等，為各種不同基礎模型提供統(tǒng)一接口
比如通過API完成一次問答
import os
os.environ["OPENAI_API_KEY"] = '你的api key'
from langchain.llms import OpenAI
llm = OpenAI(model_name="text-davinci-003",max_tokens=1024)
llm("怎么評價人工智能")得到的回答如下圖所示
1.1.2 LLMS層
這一層主要強調對models層能力的封裝以及服務化輸出能力，主要有：
- 各類LLM模型管理平臺：強調的模型的種類豐富度以及易用性
- 一體化服務能力產品：強調開箱即用
- 差異化能力：比如聚焦于Prompt管理(包括提示管理、提示優(yōu)化和提示序列化)、基于共享資源的模型運行模式等等
比如Google's PaLM Text APIs，再比如 llms/openai.py?文件下
model_token_mapping = {
"gpt-4": 8192,
"gpt-4-0314": 8192,
"gpt-4-0613": 8192,
"gpt-4-32k": 32768,
"gpt-4-32k-0314": 32768,
"gpt-4-32k-0613": 32768,
"gpt-3.5-turbo": 4096,
"gpt-3.5-turbo-0301": 4096,
"gpt-3.5-turbo-0613": 4096,
"gpt-3.5-turbo-16k": 16385,
"gpt-3.5-turbo-16k-0613": 16385,
"text-ada-001": 2049,
"ada": 2049,
"text-babbage-001": 2040,
"babbage": 2049,
"text-curie-001": 2049,
"curie": 2049,
"davinci": 2049,
"text-davinci-003": 4097,
"text-davinci-002": 4097,
"code-davinci-002": 8001,
"code-davinci-001": 8001,
"code-cushman-002": 2048,
"code-cushman-001": 2048,
}1.1.3?Index(索引)：Vector方案、KG方案
對用戶私域文本、圖片、PDF等各類文檔進行存儲和檢索(相當于結構化文檔，以便讓外部數據和模型交互)，具體實現上有兩個方案：一個Vector方案、一個KG方案
1.1.3.1?Index(索引)之Vector方案
對于Vector方案：即對文件先切分為Chunks，在按Chunks分別編碼存儲并檢索，可參考此代碼文件：langchain/libs/langchain/langchain/indexes /vectorstore.py
該代碼文件依次實現
模塊導入：導入了各種類型檢查、數據結構、預定義類和函數
接下來，實現了一個函數_get_default_text_splitter，兩個類VectorStoreIndexWrapper、VectorstoreIndexCreator
_get_default_text_splitter 函數：
這是一個私有函數，返回一個默認的文本分割器，它可以將文本遞歸地分割成大小為1000的塊，且塊與塊之間有重疊
# 默認的文本分割器函數
def _get_default_text_splitter() -> TextSplitter:
return RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=0)為什么要進行切割？
原因很簡單， embedding(text2vec，文本轉化為向量)以及 LLM encoder 對輸入 tokens 都有限制。embedding 會將一個 text(長字符串)的語義信息壓縮成一個向量，但其對 text 包含的 tokens 是有限制的，一段話壓縮成一個向量是 ok，但一本書壓縮成一個向量可能就丟失了絕大多數語義
接下來是，VectorStoreIndexWrapper 類：
這是一個包裝類，主要是為了方便地訪問和查詢向量存儲（Vector Store）
- vectorstore: 一個向量存儲對象的屬性
vectorstore: VectorStore # 向量存儲對象 class Config: """Configuration for this pydantic object.""" extra = Extra.forbid # 額外配置項 arbitrary_types_allowed = True # 允許任意類型 - query: 一個方法，它接受一個問題字符串并查詢向量存儲來獲取答案
解釋一下上面出現的提取器
# 查詢向量存儲的函數 def query( self, question: str, # 輸入的問題字符串 llm: Optional[BaseLanguageModel] = None, # 可選的語言模型參數，默認為None retriever_kwargs: Optional[Dict[str, Any]] = None, # 提取器的可選參數，默認為None **kwargs: Any # 其他關鍵字參數 ) -> str: """Query the vectorstore.""" # 函數的文檔字符串，描述函數的功能 # 如果沒有提供語言模型參數，則使用OpenAI作為默認語言模型，并設定溫度參數為0 llm = llm or OpenAI(temperature=0) # 如果沒有提供提取器的參數，則初始化為空字典 retriever_kwargs = retriever_kwargs or {} # 創(chuàng)建一個基于語言模型和向量存儲提取器的檢索QA鏈 chain = RetrievalQA.from_chain_type( llm, retriever=self.vectorstore.as_retriever(**retriever_kwargs), **kwargs ) # 使用創(chuàng)建的QA鏈運行提供的問題，并返回結果 return chain.run(question)提取器首先從大型語料庫中檢索與問題相關的文檔或片段，然后生成器根據這些檢索到的文檔生成答案。
提取器可以基于許多不同的技術，包括：
? ? a.基于關鍵字的檢索：使用關鍵字匹配來查找相關文檔
? ? b.向量空間模型：將文檔和查詢都表示為向量，并通過計算它們之間的相似度來檢索相關文檔
? ? c.基于深度學習的方法：使用預訓練的神經網絡模型（如BERT、RoBERTa等）將文檔和查詢編碼為向量，并進行相似度計算
? ? d.索引方法：例如倒排索引，這是搜索引擎常用的技術，可以快速找到包含特定詞或短語的文檔
這些方法可以獨立使用，也可以結合使用，以提高檢索的準確性和速度 - query_with_sources: 類似于query，但它還返回與查詢結果相關的數據源
# 查詢向量存儲并返回數據源的函數 def query_with_sources( self, question: str, llm: Optional[BaseLanguageModel] = None, retriever_kwargs: Optional[Dict[str, Any]] = None, **kwargs: Any ) -> dict: """Query the vectorstore and get back sources.""" llm = llm or OpenAI(temperature=0) # 默認使用OpenAI作為語言模型 retriever_kwargs = retriever_kwargs or {} # 提取器參數 chain = RetrievalQAWithSourcesChain.from_chain_type( llm, retriever=self.vectorstore.as_retriever(**retriever_kwargs), **kwargs ) return chain({chain.question_key: question})
最后是VectorstoreIndexCreator 類：
這是一個創(chuàng)建向量存儲索引的類
- vectorstore_cls: 使用的向量存儲類，默認為Chroma
一個簡化的向量存儲可以看作是一個大型的表格或數據庫，其中每行代表一個項目（如文檔、圖像、句子等），而每個項目則有一個與之關聯的高維向量。向量的維度可以從幾十到幾千，取決于所使用的嵌入模型
vectorstore_cls: Type[VectorStore] = Chroma # 默認使用Chroma作為向量存儲類
例如：
至于這里的Chroma是一種常見的向量數據庫，可以通過與LangChain的集成，實現基于語言模型的各種應用Item ID Vector (in a high dimensional space) 1 [0.34, -0.2, 0.5, ...] 2 [-0.1, 0.3, -0.4, ...] ... ... 
- embedding: 使用的嵌入類，默認為OpenAIEmbeddings
順帶說一下，Huggingface 有一個 embedding 的 benchmark：https://huggingface.co/spaces/mteb/leaderboard
embedding: Embeddings = Field(default_factory=OpenAIEmbeddings) # 默認使用OpenAIEmbeddings作為嵌入類 - text_splitter: 用于分割文本的文本分割器
text_splitter: TextSplitter = Field(default_factory=_get_default_text_splitter) # 默認文本分割器 - from_loaders: 從給定的加載器列表中創(chuàng)建一個向量存儲索引
# 從加載器創(chuàng)建向量存儲索引的函數 def from_loaders(self, loaders: List[BaseLoader]) -> VectorStoreIndexWrapper: """Create a vectorstore index from loaders.""" docs = [] for loader in loaders: # 遍歷加載器 docs.extend(loader.load()) # 加載文檔 return self.from_documents(docs) - from_documents: 從給定的文檔列表中創(chuàng)建一個向量存儲索引
# 從文檔創(chuàng)建向量存儲索引的函數 def from_documents(self, documents: List[Document]) -> VectorStoreIndexWrapper: """Create a vectorstore index from documents.""" sub_docs = self.text_splitter.split_documents(documents) # 分割文檔 vectorstore = self.vectorstore_cls.from_documents( sub_docs, self.embedding, **self.vectorstore_kwargs # 從文檔創(chuàng)建向量存儲 ) return VectorStoreIndexWrapper(vectorstore=vectorstore) # 返回向量存儲的包裝對象
1.1.3.2?Index(索引)之KG方案
對于KG方案：這部分利用LLM抽取文件中的三元組，將其存儲為KG供后續(xù)檢索，可參考此代碼文件：langchain/libs/langchain/langchain/indexes /graph.py
"""Graph Index Creator.""" # 定義"圖索引創(chuàng)建器"的描述
# 導入相關的模塊和類型定義
from typing import Optional, Type # 導入可選類型和類型的基礎類型
from langchain import BasePromptTemplate # 導入基礎提示模板
from langchain.chains.llm import LLMChain # 導入LLM鏈
from langchain.graphs.networkx_graph import NetworkxEntityGraph, parse_triples # 導入Networkx實體圖和解析三元組的功能
from langchain.indexes.prompts.knowledge_triplet_extraction import ( # 從知識三元組提取模塊導入對應的提示
KNOWLEDGE_TRIPLE_EXTRACTION_PROMPT,
)
from langchain.pydantic_v1 import BaseModel # 導入基礎模型
from langchain.schema.language_model import BaseLanguageModel # 導入基礎語言模型的定義
class GraphIndexCreator(BaseModel): # 定義圖索引創(chuàng)建器類，繼承自BaseModel
"""Functionality to create graph index.""" # 描述該類的功能為"創(chuàng)建圖索引"
llm: Optional[BaseLanguageModel] = None # 定義可選的語言模型屬性，默認為None
graph_type: Type[NetworkxEntityGraph] = NetworkxEntityGraph # 定義圖的類型，默認為NetworkxEntityGraph
def from_text(
self, text: str, prompt: BasePromptTemplate = KNOWLEDGE_TRIPLE_EXTRACTION_PROMPT
) -> NetworkxEntityGraph: # 定義一個方法，從文本中創(chuàng)建圖索引
"""Create graph index from text.""" # 描述該方法的功能
if self.llm is None: # 如果語言模型為None，則拋出異常
raise ValueError("llm should not be None")
graph = self.graph_type() # 創(chuàng)建一個新的圖
chain = LLMChain(llm=self.llm, prompt=prompt) # 使用當前的語言模型和提示創(chuàng)建一個LLM鏈
output = chain.predict(text=text) # 使用LLM鏈對文本進行預測
knowledge = parse_triples(output) # 解析預測輸出得到的三元組
for triple in knowledge: # 遍歷所有的三元組
graph.add_triple(triple) # 將三元組添加到圖中
return graph # 返回創(chuàng)建的圖
async def afrom_text( # 定義一個異步版本的from_text方法
self, text: str, prompt: BasePromptTemplate = KNOWLEDGE_TRIPLE_EXTRACTION_PROMPT
) -> NetworkxEntityGraph:
"""Create graph index from text asynchronously.""" # 描述該異步方法的功能
if self.llm is None: # 如果語言模型為None，則拋出異常
raise ValueError("llm should not be None")
graph = self.graph_type() # 創(chuàng)建一個新的圖
chain = LLMChain(llm=self.llm, prompt=prompt) # 使用當前的語言模型和提示創(chuàng)建一個LLM鏈
output = await chain.apredict(text=text) # 異步使用LLM鏈對文本進行預測
knowledge = parse_triples(output) # 解析預測輸出得到的三元組
for triple in knowledge: # 遍歷所有的三元組
graph.add_triple(triple) # 將三元組添加到圖中
return graph # 返回創(chuàng)建的圖另外，為了索引，便不得不牽涉以下這些能力
-
Document Loaders，文檔加載的標準接口
與各種格式的文檔及數據源集成，比如Arxiv、Email、Excel、Markdown、PDF(所以可以做類似ChatPDF這樣的應用)、Youtube …
相近的還有
docstore，其中包含wikipedia.py等
document_transformers -
embeddings?(langchain/libs/langchain/langchain/embeddings)，則涉及到各種embeddings算法，分別體現在各種代碼文件中：
elasticsearch.py、google_palm.py、gpt4all.py、huggingface.py、huggingface_hub.py
llamacpp.py、minimax.py、modelscope_hub.py、mosaicml.py
openai.py
sentence_transformer.py、spacy_embeddings.py、tensorflow_hub.py、vertexai.py
1.2 能力層：Chains、Memory、Tools
如果基礎層提供了最核心的能力，能力層則給這些能力安裝上手、腳、腦，讓其具有記憶和觸發(fā)萬物的能力，包括：Chains、Memory、Tool三部分
1.2.1?Chains：鏈接
簡言之，相當于包括一系列對各種組件的調用，可能是一個 Prompt 模板，一個語言模型，一個輸出解析器，一起工作處理用戶的輸入，生成響應，并處理輸出
具體而言，則相當于按照不同的需求抽象并定制化不同的執(zhí)行邏輯，Chain可以相互嵌套并串行執(zhí)行，通過這一層，讓LLM的能力鏈接到各行各業(yè)
- 比如與Elasticsearch數據庫交互的：elasticsearch_database
- 比如基于知識圖譜問答的：graph_qa
?其中的代碼文件：chains/graph_qa/base.py 便實現了一個基于知識圖譜實現的問答系統(tǒng)，具體步驟為
首先，根據提取到的實體在知識圖譜中查找相關的信息「這是通過?self.graph.get_entity_knowledge(entity) 實現的，它返回的是與實體相關的所有信息，形式為三元組」
然后，將所有的三元組組合起來，形成上下文
最后，將問題和上下文一起輸入到qa_chain，得到最后的答案entities = get_entities(entity_string) # 獲取實體列表。 context = "" # 初始化上下文。 all_triplets = [] # 初始化三元組列表。 for entity in entities: # 遍歷每個實體 all_triplets.extend(self.graph.get_entity_knowledge(entity)) # 獲取實體的所有知識并加入到三元組列表中。 context = "\n".join(all_triplets) # 用換行符連接所有的三元組作為上下文。 # 打印完整的上下文。 _run_manager.on_text("Full Context:", end="\n", verbose=self.verbose) _run_manager.on_text(context, color="green", end="\n", verbose=self.verbose) # 使用上下文和問題獲取答案。 result = self.qa_chain( {"question": question, "context": context}, callbacks=_run_manager.get_child(), ) return {self.output_key: result[self.qa_chain.output_key]} # 返回答案 - 比如能自動生成代碼并執(zhí)行的：llm_math等等
- 比如面向私域數據的：qa_with_sources，其中的這份代碼文件 chains/qa_with_sources/vector_db.py 則是使用向量數據庫的問題回答，核心在于以下兩個函數
reduce_tokens_below_limit?_get_docs# 定義基于向量數據庫的問題回答類 class VectorDBQAWithSourcesChain(BaseQAWithSourcesChain): """Question-answering with sources over a vector database.""" # 定義向量數據庫的字段 vectorstore: VectorStore = Field(exclude=True) """Vector Database to connect to.""" # 定義返回結果的數量 k: int = 4 # 是否基于token限制來減少返回結果的數量 reduce_k_below_max_tokens: bool = False # 定義返回的文檔基于token的最大限制 max_tokens_limit: int = 3375 # 定義額外的搜索參數 search_kwargs: Dict[str, Any] = Field(default_factory=dict) # 定義函數來根據最大token限制來減少文檔 def _reduce_tokens_below_limit(self, docs: List[Document]) -> List[Document]: num_docs = len(docs) # 檢查是否需要根據token減少文檔數量 if self.reduce_k_below_max_tokens and isinstance( self.combine_documents_chain, StuffDocumentsChain ): tokens = [ self.combine_documents_chain.llm_chain.llm.get_num_tokens( doc.page_content ) for doc in docs ] token_count = sum(tokens[:num_docs]) # 減少文檔數量直到滿足token限制 while token_count > self.max_tokens_limit: num_docs -= 1 token_count -= tokens[num_docs] return docs[:num_docs]# 獲取相關文檔的函數 def _get_docs( self, inputs: Dict[str, Any], *, run_manager: CallbackManagerForChainRun ) -> List[Document]: question = inputs[self.question_key] # 從向量存儲中搜索相似的文檔 docs = self.vectorstore.similarity_search( question, k=self.k, **self.search_kwargs ) return self._reduce_tokens_below_limit(docs) - 比如面向SQL數據源的：sql_database，可以重點關注這份代碼文件：chains/sql_database/query.py
- 比如面向模型對話的：chat_models，包括這些代碼文件：__init__.py、anthropic.py、azure_openai.py、base.py、fake.py、google_palm.py、human.py、jinachat.py、openai.py、promptlayer_openai.py、vertexai.py
另外，還有比較讓人眼前一亮的：
constitutional_ai：對最終結果進行偏見、合規(guī)問題處理的邏輯，保證最終的結果符合價值觀
llm_checker：能讓LLM自動檢測自己的輸出是否有沒有問題的邏輯
1.2.2 Memory：記憶
簡言之，用來保存和模型交互時的上下文狀態(tài)，處理長期記憶
具體而言，這層主要有兩個核心點：
? 對Chains的執(zhí)行過程中的輸入、輸出進行記憶并結構化存儲，為下一步的交互提供上下文，這部分簡單存儲在Redis即可
? 根據交互歷史構建知識圖譜，根據關聯信息給出準確結果，對應的代碼文件為：memory/kg.py
# 定義知識圖譜對話記憶類
class ConversationKGMemory(BaseChatMemory):
"""知識圖譜對話記憶類
在對話中與外部知識圖譜集成，存儲和檢索對話中的知識三元組信息。
"""
k: int = 2 # 考慮的上下文對話數量
human_prefix: str = "Human" # 人類前綴
ai_prefix: str = "AI" # AI前綴
kg: NetworkxEntityGraph = Field(default_factory=NetworkxEntityGraph) # 知識圖譜實例
knowledge_extraction_prompt: BasePromptTemplate = KNOWLEDGE_TRIPLE_EXTRACTION_PROMPT # 知識提取提示
entity_extraction_prompt: BasePromptTemplate = ENTITY_EXTRACTION_PROMPT # 實體提取提示
llm: BaseLanguageModel # 基礎語言模型
summary_message_cls: Type[BaseMessage] = SystemMessage # 總結消息類
memory_key: str = "history" # 歷史記憶鍵
def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
"""返回歷史緩沖區(qū)。"""
entities = self._get_current_entities(inputs) # 獲取當前實體
summary_strings = []
for entity in entities: # 對于每個實體
knowledge = self.kg.get_entity_knowledge(entity) # 獲取與實體相關的知識
if knowledge:
summary = f"On {entity}: {'. '.join(knowledge)}." # 構建總結字符串
summary_strings.append(summary)
context: Union[str, List]
if not summary_strings:
context = [] if self.return_messages else ""
elif self.return_messages:
context = [
self.summary_message_cls(content=text) for text in summary_strings
]
else:
context = "\n".join(summary_strings)
return {self.memory_key: context}
@property
def memory_variables(self) -> List[str]:
"""始終返回記憶變量列表。"""
return [self.memory_key]
def _get_prompt_input_key(self, inputs: Dict[str, Any]) -> str:
"""獲取提示的輸入鍵。"""
if self.input_key is None:
return get_prompt_input_key(inputs, self.memory_variables)
return self.input_key
def _get_prompt_output_key(self, outputs: Dict[str, Any]) -> str:
"""獲取提示的輸出鍵。"""
if self.output_key is None:
if len(outputs) != 1:
raise ValueError(f"One output key expected, got {outputs.keys()}")
return list(outputs.keys())[0]
return self.output_key
def get_current_entities(self, input_string: str) -> List[str]:
"""從輸入字符串中獲取當前實體。"""
chain = LLMChain(llm=self.llm, prompt=self.entity_extraction_prompt)
buffer_string = get_buffer_string(
self.chat_memory.messages[-self.k * 2 :],
human_prefix=self.human_prefix,
ai_prefix=self.ai_prefix,
)
output = chain.predict(
history=buffer_string,
input=input_string,
)
return get_entities(output)
def _get_current_entities(self, inputs: Dict[str, Any]) -> List[str]:
"""獲取對話中的當前實體。"""
prompt_input_key = self._get_prompt_input_key(inputs)
return self.get_current_entities(inputs[prompt_input_key])
def get_knowledge_triplets(self, input_string: str) -> List[KnowledgeTriple]:
"""從輸入字符串中獲取知識三元組。"""
chain = LLMChain(llm=self.llm, prompt=self.knowledge_extraction_prompt)
buffer_string = get_buffer_string(
self.chat_memory.messages[-self.k * 2 :],
human_prefix=self.human_prefix,
ai_prefix=self.ai_prefix,
)
output = chain.predict(
history=buffer_string,
input=input_string,
verbose=True,
)
knowledge = parse_triples(output) # 解析三元組
return knowledge
def _get_and_update_kg(self, inputs: Dict[str, Any]) -> None:
"""從對話歷史中獲取并更新知識圖譜。"""
prompt_input_key = self._get_prompt_input_key(inputs)
knowledge = self.get_knowledge_triplets(inputs[prompt_input_key])
for triple in knowledge:
self.kg.add_triple(triple) # 向知識圖譜中添加三元組
def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:
"""將此對話的上下文保存到緩沖區(qū)。"""
super().save_context(inputs, outputs)
self._get_and_update_kg(inputs)
def clear(self) -> None:
"""清除記憶內容。"""
super().clear()
self.kg.clear() # 清除知識圖譜內容1.2.3 Tools層，工具
其實Chains層可以根據LLM + Prompt執(zhí)行一些特定的邏輯，但是如果要用Chain實現所有的邏輯不現實，可以通過Tools層也可以實現，Tools層理解為技能比較合理，典型的比如搜索、Wikipedia、天氣預報、ChatGPT服務等等
1.3 應用層：Agents
1.3.1?Agents：代理
簡言之，有了基礎層和能力層，我們可以構建各種各樣好玩的，有價值的服務，這里就是Agent
具體而言，Agent 作為代理人去向 LLM 發(fā)出請求，然后采取行動，且檢查結果直到工作完成，包括LLM無法處理的任務的代理 (例如搜索或計算，類似ChatGPT plus的插件有調用bing和計算器的功能)
比如，Agent 可以使用維基百科查找 Barack Obama 的出生日期，然后使用計算器計算他在 2023 年的年齡
# pip install wikipedia
from langchain.agents import load_tools
from langchain.agents import initialize_agent
from langchain.agents import AgentType
tools = load_tools(["wikipedia", "llm-math"], llm=llm)
agent = initialize_agent(tools,?
? ? ? ? ? ? ? ? ? ? ? ? ?llm,?
? ? ? ? ? ? ? ? ? ? ? ? ?agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,?
? ? ? ? ? ? ? ? ? ? ? ? ?verbose=True)
agent.run("奧巴馬的生日是哪天? 到2023年他多少歲了?")此外，關于Wikipedia可以關注下這個代碼文件：langchain/docstore/wikipedia.py?...
最終langchain的整體技術架構可以如下圖所示 (查看高清大圖，此外，這里還有另一個架構圖)
第二部分?基于LangChain + ChatGLM-6B(23年7月初版)的本地知識庫問答
2.1 核心步驟：如何通過LangChain+LLM實現本地知識庫問答
2023年7月，GitHub上有一個利用?langchain?思想實現的基于本地知識庫的問答應用：langchain-ChatGLM (這是其GitHub地址，當然還有和它類似的但現已支持Vicuna-13b的項目，比如LangChain-ChatGLM-Webui?)，目標期望建立一套對中文場景與開源模型支持友好、可離線運行的知識庫問答解決方案
- 該項目受?GanymedeNil?的項目?document.ai，和?AlexZhangji?創(chuàng)建的?ChatGLM-6B Pull Request?啟發(fā)，建立了全流程可使用開源模型實現的本地知識庫問答應用?，F已支持使用?ChatGLM-6B、?ClueAI/ChatYuan-large-v2?等大語言模型的接入
- 該項目中 Embedding 默認選用的是?GanymedeNil/text2vec-large-chinese，LLM 默認選用的是?ChatGLM-6B，依托上述模型，本項目可實現全部使用開源模型離線私有部署
本項目實現原理如下圖所示 (與基于文檔的問答 大同小異，過程包括：1 加載文檔?-> 2 讀取文檔?-> 3/4文檔分割 -> 5/6 文本向量化 -> 8/9 問句向量化 -> 10 在文檔向量中匹配出與問句向量最相似的top k個 -> 11/12/13 匹配出的文本作為上下文和問題一起添加到prompt中 -> 14/15提交給LLM生成回答?)
-
第一階段：加載文件-讀取文件-文本分割(Text splitter)
加載文件：這是讀取存儲在本地的知識庫文件的步驟
讀取文件：讀取加載的文件內容，通常是將其轉化為文本格式
文本分割(Text splitter)：按照一定的規(guī)則(例如段落、句子、詞語等)將文本分割 -
第二階段：文本向量化(embedding)-存儲到向量數據庫
文本向量化(embedding)：這通常涉及到NLP的特征抽取，可以通過諸如TF-IDF、word2vec、BERT等方法將分割好的文本轉化為數值向量
存儲到向量數據庫：文本向量化之后存儲到數據庫vectorstore (FAISS，下一節(jié)會詳解FAISS)def init_vector_store(self): persist_dir = os.path.join(VECTORE_PATH, ".vectordb") # 持久化向量數據庫的地址 print("向量數據庫持久化地址: ", persist_dir) # 打印持久化地址 # 如果持久化地址存在 if os.path.exists(persist_dir): # 從本地持久化文件中加載 print("從本地向量加載數據...") # 使用 Chroma 加載持久化的向量數據 vector_store = Chroma(persist_directory=persist_dir, embedding_function=self.embeddings) # 如果持久化地址不存在 else: # 加載知識庫 documents = self.load_knownlege() # 使用 Chroma 從文檔中創(chuàng)建向量存儲 vector_store = Chroma.from_documents(documents=documents, embedding=self.embeddings, persist_directory=persist_dir) vector_store.persist() # 持久化向量存儲 return vector_store # 返回向量存儲其中l(wèi)oad_knownlege的實現為
def load_knownlege(self): docments = [] # 初始化一個空列表來存儲文檔 # 遍歷 DATASETS_DIR 目錄下的所有文件 for root, _, files in os.walk(DATASETS_DIR, topdown=False): for file in files: filename = os.path.join(root, file) # 獲取文件的完整路徑 docs = self._load_file(filename) # 加載文件中的文檔 # 更新 metadata 數據 new_docs = [] # 初始化一個空列表來存儲新文檔 for doc in docs: # 更新文檔的 metadata，將 "source" 字段的值替換為不包含 DATASETS_DIR 的相對路徑 doc.metadata = {"source": doc.metadata["source"].replace(DATASETS_DIR, "")} print("文檔2向量初始化中, 請稍等...", doc.metadata) # 打印正在初始化的文檔的 metadata new_docs.append(doc) # 將文檔添加到新文檔列表 docments += new_docs # 將新文檔列表添加到總文檔列表 return docments # 返回所有文檔的列表 -
第三階段：問句向量化
這是將用戶的查詢或問題轉化為向量，應使用與文本向量化相同的方法，以便在相同的空間中進行比較 -
第四階段：在文本向量中匹配出與問句向量最相似的top k個
這一步是信息檢索的核心，通過計算余弦相似度、歐氏距離等方式，找出與問句向量最接近的文本向量def query(self, q): """在向量數據庫中查找與問句向量相似的文本向量""" vector_store = self.init_vector_store() docs = vector_store.similarity_search_with_score(q, k=self.top_k) for doc in docs: dc, s = doc yield s, dc -
第五階段：匹配出的文本作為上下文和問題一起添加到prompt中
這是利用匹配出的文本來形成與問題相關的上下文，用于輸入給語言模型 -
第六階段：提交給LLM生成回答
最后，將這個問題和上下文一起提交給語言模型(例如GPT系列)，讓它生成回答
比如知識查詢(代碼來源)class KnownLedgeBaseQA: # 初始化 def __init__(self) -> None: k2v = KnownLedge2Vector() # 創(chuàng)建一個知識到向量的轉換器 self.vector_store = k2v.init_vector_store() # 初始化向量存儲 self.llm = VicunaLLM() # 創(chuàng)建一個 VicunaLLM 對象 # 獲得與查詢相似的答案 def get_similar_answer(self, query): # 創(chuàng)建一個提示模板 prompt = PromptTemplate( template=conv_qa_prompt_template, input_variables=["context", "question"] # 輸入變量包括 "context"（上下文） 和 "question"（問題） ) # 使用向量存儲來檢索文檔 retriever = self.vector_store.as_retriever(search_kwargs={"k": VECTOR_SEARCH_TOP_K}) docs = retriever.get_relevant_documents(query=query) # 獲取與查詢相關的文本 context = [d.page_content for d in docs] # 從文本中提取出內容 result = prompt.format(context="\n".join(context), question=query) # 格式化模板，并用從文本中提取出的內容和問題填充 return result # 返回結果
如你所見，這種通過組合langchain+LLM的方式，特別適合一些垂直領域或大型集團企業(yè)搭建通過LLM的智能對話能力搭建企業(yè)內部的私有問答系統(tǒng)，也適合個人專門針對一些英文paper進行問答，比如比較火的一個開源項目：ChatPDF，其從文檔處理角度來看，實現流程如下(圖源)：
2.2?Facebook AI Similarity Search(FAISS)：高效向量相似度檢索
Faiss的全稱是Facebook AI Similarity Search (官方介紹頁、GitHub地址)，是FaceBook的AI團隊針對大規(guī)模相似度檢索問題開發(fā)的一個工具，使用C++編寫，有python接口，對10億量級的索引可以做到毫秒級檢索的性能
簡單來說，Faiss的工作，就是把我們自己的候選向量集封裝成一個index數據庫，它可以加速我們檢索相似向量TopK的過程，其中有些索引還支持GPU構建
2.2.1?Faiss檢索相似向量TopK的基本流程
Faiss檢索相似向量TopK的工程基本都能分為三步：
- 得到向量庫
import numpy as np d = 64 # 向量維度 nb = 100000 # index向量庫的數據量 nq = 10000 # 待檢索query的數目 np.random.seed(1234) xb = np.random.random((nb, d)).astype('float32') xb[:, 0] += np.arange(nb) / 1000. # index向量庫的向量 xq = np.random.random((nq, d)).astype('float32') xq[:, 0] += np.arange(nq) / 1000. # 待檢索的query向量 - 用faiss 構建index，并將向量添加到index中
其中的構建索引選用暴力檢索的方法FlatL2，L2代表構建的index采用的相似度度量方法為L2范數，即歐氏距離import faiss index = faiss.IndexFlatL2(d) print(index.is_trained) # 輸出為True，代表該類index不需要訓練，只需要add向量進去即可 index.add(xb) # 將向量庫中的向量加入到index中 print(index.ntotal) # 輸出index中包含的向量總數，為100000 - 用faiss index 檢索，檢索出TopK的相似query
k = 4 # topK的K值 D, I = index.search(xq, k)# xq為待檢索向量，返回的I為每個待檢索query最相似TopK的索引list，D為其對應的距離 print(I[:5]) print(D[-5:])打印輸出為：
>>>?
[[ ?0 393 363 ?78]?
?[ ?1 555 277 364]?
?[ ?2 304 101 ?13]?
?[ ?3 173 ?18 182]?
?[ ?4 288 370 531]] ?
[[ 0. ? ? ? ? ?7.17517328 ?7.2076292 ? 7.25116253] ?
?[ 0. ? ? ? ? ?6.32356453 ?6.6845808 ? 6.79994535] ?
?[ 0. ? ? ? ? ?5.79640865 ?6.39173603 ?7.28151226] ?
?[ 0. ? ? ? ? ?7.27790546 ?7.52798653 ?7.66284657] ?
?[ 0. ? ? ? ? ?6.76380348 ?7.29512024 ?7.36881447]]
2.2.2?FAISS構建索引的多種方式
構建index方法和傳參方法可以為
dim, measure = 64, faiss.METRIC_L2
param = 'Flat'
index = faiss.index_factory(dim, param, measure)- dim為向量維數
- 最重要的是param參數，它是傳入index的參數，代表需要構建什么類型的索引；
- measure為度量方法，目前支持兩種，歐氏距離和inner product，即內積。因此，要計算余弦相似度，只需要將vecs歸一化后，使用內積度量即可
此文，現在faiss官方支持八種度量方式，分別是：
- METRIC_INNER_PRODUCT（內積）
- METRIC_L1（曼哈頓距離）
- METRIC_L2（歐氏距離）
- METRIC_Linf（無窮范數）
- METRIC_Lp（p范數）
- METRIC_BrayCurtis（BC相異度）
- METRIC_Canberra（蘭氏距離/堪培拉距離）
- METRIC_JensenShannon（JS散度）
2.2.2.1 Flat ：暴力檢索
- 優(yōu)點：該方法是Faiss所有index中最準確的，召回率最高的方法，沒有之一；
- 缺點：速度慢，占內存大。
- 使用情況：向量候選集很少，在50萬以內，并且內存不緊張。
- 注：雖然都是暴力檢索，faiss的暴力檢索速度比一般程序猿自己寫的暴力檢索要快上不少，所以并不代表其無用武之地，建議有暴力檢索需求的同學還是用下faiss。
- 構建方法：
dim, measure = 64, faiss.METRIC_L2
param = 'Flat'
index = faiss.index_factory(dim, param, measure)
index.is_trained # 輸出為True
index.add(xb) # 向index中添加向量2.2.2.2 IVFx Flat ：倒排暴力檢索
- 優(yōu)點：IVF主要利用倒排的思想，在文檔檢索場景下的倒排技術是指，一個kw后面掛上很多個包含該詞的doc，由于kw數量遠遠小于doc，因此會大大減少了檢索的時間。在向量中如何使用倒排呢？可以拿出每個聚類中心下的向量ID，每個中心ID后面掛上一堆非中心向量，每次查詢向量的時候找到最近的幾個中心ID，分別搜索這幾個中心下的非中心向量。通過減小搜索范圍，提升搜索效率。
- 缺點：速度也還不是很快。
- 使用情況：相比Flat會大大增加檢索的速度，建議百萬級別向量可以使用。
- 參數：IVFx中的x是k-means聚類中心的個數
- 構建方法：
dim, measure = 64, faiss.METRIC_L2
param = 'IVF100,Flat' # 代表k-means聚類中心為100,
index = faiss.index_factory(dim, param, measure)
print(index.is_trained) # 此時輸出為False，因為倒排索引需要訓練k-means，
index.train(xb) # 因此需要先訓練index，再add向量
index.add(xb) 2.2.2.3 PQx ：乘積量化
- 優(yōu)點：利用乘積量化的方法，改進了普通檢索，將一個向量的維度切成x段，每段分別進行檢索，每段向量的檢索結果取交集后得出最后的TopK。因此速度很快，而且占用內存較小，召回率也相對較高。
- 缺點：召回率相較于暴力檢索，下降較多。
- 使用情況：內存及其稀缺，并且需要較快的檢索速度，不那么在意召回率
- 參數：PQx中的x為將向量切分的段數，因此，x需要能被向量維度整除，且x越大，切分越細致，時間復雜度越高
- 構建方法：
dim, measure = 64, faiss.METRIC_L2
param = 'PQ16'
index = faiss.index_factory(dim, param, measure)
print(index.is_trained) # 此時輸出為False，因為倒排索引需要訓練k-means，
index.train(xb) # 因此需要先訓練index，再add向量
index.add(xb) 2.2.2.4 IVFxPQy 倒排乘積量化
- 優(yōu)點：工業(yè)界大量使用此方法，各項指標都均可以接受，利用乘積量化的方法，改進了IVF的k-means，將一個向量的維度切成x段，每段分別進行k-means再檢索。
- 缺點：集百家之長，自然也集百家之短
- 使用情況：一般來說，各方面沒啥特殊的極端要求的話，最推薦使用該方法！
- 參數：IVFx，PQy，其中的x和y同上
- 構建方法：
dim, measure = 64, faiss.METRIC_L2
param = 'IVF100,PQ16'
index = faiss.index_factory(dim, param, measure)
print(index.is_trained) # 此時輸出為False，因為倒排索引需要訓練k-means，
index.train(xb) # 因此需要先訓練index，再add向量 index.add(xb) 2.2.2.5 LSH 局部敏感哈希
- 原理：哈希對大家再熟悉不過，向量也可以采用哈希來加速查找，我們這里說的哈希指的是局部敏感哈希（Locality Sensitive Hashing，LSH），不同于傳統(tǒng)哈希盡量不產生碰撞，局部敏感哈希依賴碰撞來查找近鄰。高維空間的兩點若距離很近，那么設計一種哈希函數對這兩點進行哈希計算后分桶，使得他們哈希分桶值有很大的概率是一樣的，若兩點之間的距離較遠，則他們哈希分桶值相同的概率會很小。
- 優(yōu)點：訓練非?？?，支持分批導入，index占內存很小，檢索也比較快
- 缺點：召回率非常拉垮。
- 使用情況：候選向量庫非常大，離線檢索，內存資源比較稀缺的情況
- 構建方法：
dim, measure = 64, faiss.METRIC_L2
param = 'LSH'
index = faiss.index_factory(dim, param, measure)
print(index.is_trained) # 此時輸出為True
index.add(xb) 2.2.2.6 HNSWx
- 優(yōu)點：該方法為基于圖檢索的改進方法，檢索速度極快，10億級別秒出檢索結果，而且召回率幾乎可以媲美Flat，最高能達到驚人的97%。檢索的時間復雜度為loglogn，幾乎可以無視候選向量的量級了。并且支持分批導入，極其適合線上任務，毫秒級別體驗。
- 缺點：構建索引極慢，占用內存極大（是Faiss中最大的，大于原向量占用的內存大?。?/li>
- 參數：HNSWx中的x為構建圖時每個點最多連接多少個節(jié)點，x越大，構圖越復雜，查詢越精確，當然構建index時間也就越慢，x取4~64中的任何一個整數。
- 使用情況：不在乎內存，并且有充裕的時間來構建index
- 構建方法：
dim, measure = 64, faiss.METRIC_L2
param = 'HNSW64'
index = faiss.index_factory(dim, param, measure)
print(index.is_trained) # 此時輸出為True
index.add(xb)2.3 項目部署：langchain + ChatGLM-6B搭建本地知識庫問答
2.3.1 部署過程一：支持多種使用模式
其中的LLM模型可以根據實際業(yè)務的需求選定，本項目中用的ChatGLM-6B，其GitHub地址為：https://github.com/THUDM/ChatGLM-6B
ChatGLM-6B 是?個開源的、?持中英雙語的對話語?模型，基于 General LanguageModel (GLM) 架構，具有 62 億參數。結合模型量化技術，用戶可以在消費級的顯卡上進行本地部署（INT4 量化級別下最低只需 6GB 顯存）
ChatGLM-6B 使用了和 ChatGPT 相似的技術，針對中文問答和對話進行了優(yōu)化。經過約 1T 標識符的中英雙語訓練，輔以監(jiān)督微調、反饋自助、人類反饋強化學習等技術的加持，62 億參數的 ChatGLM-6B 已經能生成相當符合人類偏好的回答
- 新建一個python3.8.13的環(huán)境（模型文件還是可以用的）
conda create -n langchain python==3.8.13 - 拉取項目
git clone https://github.com/imClumsyPanda/langchain-ChatGLM.git - 進入目錄
cd langchain-ChatGLM - 安裝requirements.txt
conda activate langchain pip install -r requirements.txt - 當前環(huán)境支持裝langchain的最高版本是0.0.166，無法安裝0.0.174，就先裝下0.0.166試下
修改配置文件路徑：vi configs/model_config.py - 將chatglm-6b的路徑設置成自己的
“chatglm-6b”: { “name”: “chatglm-6b”, “pretrained_model_name”: “/data/sim_chatgpt/chatglm-6b”, “l(fā)ocal_model_path”: None, “provides”: “ChatGLM” - 修改要運行的代碼文件：webui.py
vi webui.py - 將最后launch函數中的share設置為True，inbrowser設置為True
- 執(zhí)行webui.py文件
可能是網絡問題，無法創(chuàng)建一個公用鏈接?？梢赃M行云服務器和本地端口的映射，參考：https://www.cnblogs.com/monologuesmw/p/14465117.html
python webui.py
對應輸出：
?占用顯存情況：大約15個G
2.3.2 部署過程二：支持多種社區(qū)上的在線體驗
項目地址：https://github.com/thomas-yanxin/LangChain-ChatGLM-Webui
HUggingFace社區(qū)在線體驗：https://huggingface.co/spaces/thomas-yanxin/LangChain-ChatLLM
另外也支持ModelScope魔搭社區(qū)、飛槳AIStudio社區(qū)等在線體驗
- 下載項目
git clone https://github.com/thomas-yanxin/LangChain-ChatGLM-Webui.git - 進入目錄
cd LangChain-ChatGLM-Webui - 安裝所需的包
pip install -r requirements.txt pip install gradio==3.10 - 修改config.py
init_llm = "ChatGLM-6B" llm_model_dict = { ? ? "chatglm": { ? ? ? ? "ChatGLM-6B": "/data/sim_chatgpt/chatglm-6b", - 修改app.py文件，將launch函數中的share設置為True，inbrowser設置為True
執(zhí)行webui.py文件python webui.py
?顯存占用約13G
第三部分 逐行深入分析：langchain-ChatGLM(23年7月初版)項目的源碼解讀
再回顧一遍langchain-ChatGLM這個項目的架構圖(圖源)
?你會發(fā)現該項目主要由以下各大模塊組成(注意，該項目的最新版已經變化很大，本第三部分可以認為是針對v0.1.14左右的版本，往上最多到v0.1.16，新版對很多功能做了更高的封裝，而從原理理解的角度來說，看老版 更好理解些)
- chains： 工作鏈路實現，如 chains/local_doc_qa 實現了基于本地?檔的問答實現
- configs：配置文件存儲
- knowledge_base/content：用于存儲上傳的原始?件
- loader： 文檔加載器的實現類
- models： llm的接?類與實現類，針對開源模型提供流式輸出?持
- textsplitter： 文本切分的實現類
- vectorstores：用于存儲向量庫?件，即本地知識庫本體
- ..
接下來，為方便讀者一目了然，更快理解
- 我基本給“下面該項目中的每一行代碼”都添加上了中文注釋
- 且為理解更順暢，我解讀各個代碼文件夾的順序是根據項目流程逐一展開的 (而非上圖GitHub上各個代碼文件夾的呈現順序)
如有問題，可以隨時留言評論
3.1?agent：custom_agent/bing_search
3.1.1 agent/custom_agent.py
from langchain.agents import Tool # 導入工具模塊
from langchain.tools import BaseTool # 導入基礎工具類
from langchain import PromptTemplate, LLMChain # 導入提示模板和語言模型鏈
from agent.custom_search import DeepSearch # 導入自定義搜索模塊
# 導入基礎單動作代理，輸出解析器，語言模型單動作代理和代理執(zhí)行器
from langchain.agents import BaseSingleActionAgent, AgentOutputParser, LLMSingleActionAgent, AgentExecutor
from typing import List, Tuple, Any, Union, Optional, Type # 導入類型注釋模塊
from langchain.schema import AgentAction, AgentFinish # 導入代理動作和代理完成模式
from langchain.prompts import StringPromptTemplate # 導入字符串提示模板
from langchain.callbacks.manager import CallbackManagerForToolRun # 導入工具運行回調管理器
from langchain.base_language import BaseLanguageModel # 導入基礎語言模型
import re # 導入正則表達式模塊
# 定義一個代理模板字符串
agent_template = """
你現在是一個{role}。這里是一些已知信息：
{related_content}
{background_infomation}
{question_guide}：{input}
{answer_format}
"""
# 定義一個自定義提示模板類，繼承自字符串提示模板
class CustomPromptTemplate(StringPromptTemplate):
template: str # 提示模板字符串
tools: List[Tool] # 工具列表
# 定義一個格式化函數，根據提供的參數生成最終的提示模板
def format(self, **kwargs) -> str:
intermediate_steps = kwargs.pop("intermediate_steps")
# 判斷是否有互聯網查詢信息
if len(intermediate_steps) == 0:
# 如果沒有，則給出默認的背景信息，角色，問題指導和回答格式
background_infomation = "\n"
role = "傻瓜機器人"
question_guide = "我現在有一個問題"
answer_format = "如果你知道答案，請直接給出你的回答！如果你不知道答案，請你只回答\"DeepSearch('搜索詞')\"，并將'搜索詞'替換為你認為需要搜索的關鍵詞，除此之外不要回答其他任何內容。\n\n下面請回答我上面提出的問題！"
else:
# 否則，根據 intermediate_steps 中的 AgentAction 拼裝 background_infomation
background_infomation = "\n\n你還有這些已知信息作為參考：\n\n"
action, observation = intermediate_steps[0]
background_infomation += f"{observation}\n"
role = "聰明的 AI 助手"
question_guide = "請根據這些已知信息回答我的問題"
answer_format = ""
kwargs["background_infomation"] = background_infomation
kwargs["role"] = role
kwargs["question_guide"] = question_guide
kwargs["answer_format"] = answer_format
return self.template.format(**kwargs) # 格式化模板并返回
# 定義一個自定義搜索工具類，繼承自基礎工具類
class CustomSearchTool(BaseTool):
name: str = "DeepSearch" # 工具名稱
description: str = "" # 工具描述
# 定義一個運行函數，接受一個查詢字符串和一個可選的回調管理器作為參數，返回DeepSearch的搜索結果
def _run(self, query: str, run_manager: Optional[CallbackManagerForToolRun] = None):
return DeepSearch.search(query = query)
# 定義一個異步運行函數，但由于DeepSearch不支持異步，所以直接拋出一個未實現錯誤
async def _arun(self, query: str):
raise NotImplementedError("DeepSearch does not support async")
# 定義一個自定義代理類，繼承自基礎單動作代理
class CustomAgent(BaseSingleActionAgent):
# 定義一個輸入鍵的屬性
@property
def input_keys(self):
return ["input"]
# 定義一個計劃函數，接受一組中間步驟和其他參數，返回一個代理動作或者代理完成
def plan(self, intermedate_steps: List[Tuple[AgentAction, str]],
**kwargs: Any) -> Union[AgentAction, AgentFinish]:
return AgentAction(tool="DeepSearch", tool_input=kwargs["input"], log="")
# 定義一個自定義輸出解析器，繼承自代理輸出解析器
class CustomOutputParser(AgentOutputParser):
# 定義一個解析函數，接受一個語言模型的輸出字符串，返回一個代理動作或者代理完成
def parse(self, llm_output: str) -> Union[AgentAction, AgentFinish]:
# 使用正則表達式匹配輸出字符串，group1是調用函數名字，group2是傳入參數
match = re.match(r'^[\s\w]*(DeepSearch)\(([^\)]+)\)', llm_output, re.DOTALL)
print(match)
# 如果語言模型沒有返回 DeepSearch() 則認為直接結束指令
if not match:
return AgentFinish(
return_values={"output": llm_output.strip()},
log=llm_output,
)
# 否則的話都認為需要調用 Tool
else:
action = match.group(1).strip()
action_input = match.group(2).strip()
return AgentAction(tool=action, tool_input=action_input.strip(" ").strip('"'), log=llm_output)
# 定義一個深度代理類
class DeepAgent:
tool_name: str = "DeepSearch" # 工具名稱
agent_executor: any # 代理執(zhí)行器
tools: List[Tool] # 工具列表
llm_chain: any # 語言模型鏈
# 定義一個查詢函數，接受一個相關內容字符串和一個查詢字符串，返回執(zhí)行器的運行結果
def query(self, related_content: str = "", query: str = ""):
tool_name =這段代碼的主要目的是建立一個深度搜索的AI代理。AI代理首先通過接收一個問題輸入，然后根據輸入生成一個提示模板，然后通過該模板引導AI生成回答或進行更深入的搜索。現在，我將繼續(xù)為剩余的代碼添加中文注釋
```python
self.tool_name
result = self.agent_executor.run(related_content=related_content, input=query ,tool_name=self.tool_name)
return result # 返回執(zhí)行器的運行結果
# 在初始化函數中，首先從DeepSearch工具創(chuàng)建一個工具實例，并添加到工具列表中
def __init__(self, llm: BaseLanguageModel, **kwargs):
tools = [
Tool.from_function(
func=DeepSearch.search,
name="DeepSearch",
description=""
)
]
self.tools = tools # 保存工具列表
tool_names = [tool.name for tool in tools] # 提取工具列表中的工具名稱
output_parser = CustomOutputParser() # 創(chuàng)建一個自定義輸出解析器實例
# 創(chuàng)建一個自定義提示模板實例
prompt = CustomPromptTemplate(template=agent_template,
tools=tools,
input_variables=["related_content","tool_name", "input", "intermediate_steps"])
# 創(chuàng)建一個語言模型鏈實例
llm_chain = LLMChain(llm=llm, prompt=prompt)
self.llm_chain = llm_chain # 保存語言模型鏈實例
# 創(chuàng)建一個語言模型單動作代理實例
agent = LLMSingleActionAgent(
llm_chain=llm_chain,
output_parser=output_parser,
stop=["\nObservation:"],
allowed_tools=tool_names
)
# 創(chuàng)建一個代理執(zhí)行器實例
agent_executor = AgentExecutor.from_agent_and_tools(agent=agent, tools=tools, verbose=True)
self.agent_executor = agent_executor # 保存代理執(zhí)行器實例3.1.2?agent/bing_search.py
#coding=utf8
# 聲明文件編碼格式為 utf8
from langchain.utilities import BingSearchAPIWrapper
# 導入 BingSearchAPIWrapper 類，這個類用于與 Bing 搜索 API 進行交互
from configs.model_config import BING_SEARCH_URL, BING_SUBSCRIPTION_KEY
# 導入配置文件中的 Bing 搜索 URL 和 Bing 訂閱密鑰
def bing_search(text, result_len=3):
# 定義一個名為 bing_search 的函數，該函數接收一個文本和結果長度的參數，默認結果長度為3
if not (BING_SEARCH_URL and BING_SUBSCRIPTION_KEY):
# 如果 Bing 搜索 URL 或 Bing 訂閱密鑰未設置，則返回一個錯誤信息的文檔
return [{"snippet": "please set BING_SUBSCRIPTION_KEY and BING_SEARCH_URL in os ENV",
"title": "env inof not fould",
"link": "https://python.langchain.com/en/latest/modules/agents/tools/examples/bing_search.html"}]
search = BingSearchAPIWrapper(bing_subscription_key=BING_SUBSCRIPTION_KEY,
bing_search_url=BING_SEARCH_URL)
# 創(chuàng)建 BingSearchAPIWrapper 類的實例，該實例用于與 Bing 搜索 API 進行交互
return search.results(text, result_len)
# 返回搜索結果，結果的數量由 result_len 參數決定
if __name__ == "__main__":
# 如果這個文件被直接運行，而不是被導入作為模塊，那么就執(zhí)行以下代碼
r = bing_search('python')
# 使用 Bing 搜索 API 來搜索 "python" 這個詞，并將結果保存在變量 r 中
print(r)
# 打印出搜索結果3.2 models：包含models和文檔加載器loader
- models： llm的接?類與實現類，針對開源模型提供流式輸出?持
- loader： 文檔加載器的實現類
3.2.1?models/chatglm_llm.py
from abc import ABC # 導入抽象基類
from langchain.llms.base import LLM # 導入語言學習模型基類
from typing import Optional, List # 導入類型標注模塊
from models.loader import LoaderCheckPoint # 導入模型加載點
from models.base import (BaseAnswer, # 導入基本回答模型
AnswerResult) # 導入回答結果模型
class ChatGLM(BaseAnswer, LLM, ABC): # 定義ChatGLM類，繼承基礎回答、語言學習模型和抽象基類
max_token: int = 10000 # 最大的token數
temperature: float = 0.01 # 溫度參數，用于控制生成文本的隨機性
top_p = 0.9 # 排序前0.9的token會被保留
checkPoint: LoaderCheckPoint = None # 檢查點模型
# history = [] # 歷史記錄
history_len: int = 10 # 歷史記錄長度
def __init__(self, checkPoint: LoaderCheckPoint = None): # 初始化方法
super().__init__() # 調用父類的初始化方法
self.checkPoint = checkPoint # 賦值檢查點模型
@property
def _llm_type(self) -> str: # 定義只讀屬性_llm_type，返回語言學習模型的類型
return "ChatGLM"
@property
def _check_point(self) -> LoaderCheckPoint: # 定義只讀屬性_check_point，返回檢查點模型
return self.checkPoint
@property
def _history_len(self) -> int: # 定義只讀屬性_history_len，返回歷史記錄的長度
return self.history_len
def set_history_len(self, history_len: int = 10) -> None: # 設置歷史記錄長度
self.history_len = history_len
def _call(self, prompt: str, stop: Optional[List[str]] = None) -> str: # 定義_call方法，實現模型的具體調用
print(f"__call:{prompt}") # 打印調用的提示信息
response, _ = self.checkPoint.model.chat( # 調用模型的chat方法，獲取回答和其他信息
self.checkPoint.tokenizer, # 使用的分詞器
prompt, # 提示信息
history=[], # 歷史記錄
max_length=self.max_token, # 最大長度
temperature=self.temperature # 溫度參數
)
print(f"response:{response}") # 打印回答信息
print(f"+++++++++++++++++++++++++++++++++++") # 打印分隔線
return response # 返回回答
def generatorAnswer(self, prompt: str,
history: List[List[str]] = [],
streaming: bool = False): # 定義生成回答的方法，可以處理流式輸入
if streaming: # 如果是流式輸入
history += [[]] # 在歷史記錄中添加新的空列表
for inum, (stream_resp, _) in enumerate(self.checkPoint.model.stream_chat( # 對模型的stream_chat方法返回的結果進行枚舉
self.checkPoint.tokenizer, # 使用的分詞器
prompt, # 提示信息
history=history[-self.history_len:-1] if self.history_len > 1 else [], # 使用的歷史記錄
max_length=self.max_token, # 最大長度
temperature=self.temperature # 溫度參數
)):
# self.checkPoint.clear_torch_cache() # 清空緩存
history[-1] = [prompt, stream_resp] # 更新最后一個歷史記錄
answer_result = AnswerResult() # 創(chuàng)建回答結果對象
answer_result.history = history # 更新回答結果的歷史記錄
answer_result.llm_output = {"answer": stream_resp} # 更新回答結果的輸出
yield answer_result # 生成回答結果
else: # 如果不是流式輸入
response, _ = self.checkPoint.model.chat( # 調用模型的chat方法，獲取回答和其他信息
self.checkPoint.tokenizer, # 使用的分詞器
prompt, # 提示信息
history=history[-self.history_len:] if self.history_len > 0 else [], # 使用的歷史記錄
max_length=self.max_token, # 最大長度
temperature=self.temperature # 溫度參數
)
self.checkPoint.clear_torch_cache() # 清空緩存
history += [[prompt, response]] # 更新歷史記錄
answer_result = AnswerResult() # 創(chuàng)建回答結果對象
answer_result.history = history # 更新回答結果的歷史記錄
answer_result.llm_output = {"answer": response} # 更新回答結果的輸出
yield answer_result # 生成回答結果3.2.2?models/shared.py
這個文件的作用是遠程調用LLM
import sys # 導入sys模塊，通常用于與Python解釋器進行交互
from typing import Any # 從typing模塊導入Any，用于表示任何類型
# 從models.loader.args模塊導入parser，可能是解析命令行參數用
from models.loader.args import parser
# 從models.loader模塊導入LoaderCheckPoint，可能是模型加載點
from models.loader import LoaderCheckPoint
# 從configs.model_config模塊導入llm_model_dict和LLM_MODEL
from configs.model_config import (llm_model_dict, LLM_MODEL)
# 從models.base模塊導入BaseAnswer，即模型的基礎類
from models.base import BaseAnswer
# 定義一個名為loaderCheckPoint的變量，類型為LoaderCheckPoint，并初始化為None
loaderCheckPoint: LoaderCheckPoint = None
def loaderLLM(llm_model: str = None, no_remote_model: bool = False, use_ptuning_v2: bool = False) -> Any:
"""
初始化 llm_model_ins LLM
:param llm_model: 模型名稱
:param no_remote_model: 是否使用遠程模型，如果需要加載本地模型，則添加 `--no-remote-model
:param use_ptuning_v2: 是否使用 p-tuning-v2 PrefixEncoder
:return:
"""
pre_model_name = loaderCheckPoint.model_name # 獲取loaderCheckPoint的模型名稱
llm_model_info = llm_model_dict[pre_model_name] # 從模型字典中獲取模型信息
if no_remote_model: # 如果不使用遠程模型
loaderCheckPoint.no_remote_model = no_remote_model # 將loaderCheckPoint的no_remote_model設置為True
if use_ptuning_v2: # 如果使用p-tuning-v2
loaderCheckPoint.use_ptuning_v2 = use_ptuning_v2 # 將loaderCheckPoint的use_ptuning_v2設置為True
if llm_model: # 如果指定了模型名稱
llm_model_info = llm_model_dict[llm_model] # 從模型字典中獲取指定的模型信息
if loaderCheckPoint.no_remote_model: # 如果不使用遠程模型
loaderCheckPoint.model_name = llm_model_info['name'] # 將loaderCheckPoint的模型名稱設置為模型信息中的name
else: # 如果使用遠程模型
loaderCheckPoint.model_name = llm_model_info['pretrained_model_name'] # 將loaderCheckPoint的模型名稱設置為模型信息中的pretrained_model_name
loaderCheckPoint.model_path = llm_model_info["local_model_path"] # 設置模型的本地路徑
if 'FastChatOpenAILLM' in llm_model_info["provides"]: # 如果模型信息中的provides包含'FastChatOpenAILLM'
loaderCheckPoint.unload_model() # 卸載模型
else: # 如果不包含
loaderCheckPoint.reload_model() # 重新加載模型
provides_class = getattr(sys.modules['models'], llm_model_info['provides']) # 獲取模型類
modelInsLLM = provides_class(checkPoint=loaderCheckPoint) # 創(chuàng)建模型實例
if 'FastChatOpenAILLM' in llm_model_info["provides"]: # 如果模型信息中的provides包含'FastChatOpenAILLM'
modelInsLLM.set_api_base_url(llm_model_info['api_base_url']) # 設置API基礎URL
modelInsLLM.call_model_name(llm_model_info['name']) # 設置模型名稱
return modelInsLLM # 返回模型實例// 待更..
3.3?configs：配置文件存儲model_config.py
import torch.cuda
import torch.backends
import os
import logging
import uuid
LOG_FORMAT = "%(levelname) -5s %(asctime)s" "-1d: %(message)s"
logger = logging.getLogger()
logger.setLevel(logging.INFO)
logging.basicConfig(format=LOG_FORMAT)
# 在以下字典中修改屬性值，以指定本地embedding模型存儲位置
# 如將 "text2vec": "GanymedeNil/text2vec-large-chinese" 修改為 "text2vec": "User/Downloads/text2vec-large-chinese"
# 此處請寫絕對路徑
embedding_model_dict = {
"ernie-tiny": "nghuyong/ernie-3.0-nano-zh",
"ernie-base": "nghuyong/ernie-3.0-base-zh",
"text2vec-base": "shibing624/text2vec-base-chinese",
"text2vec": "GanymedeNil/text2vec-large-chinese",
"m3e-small": "moka-ai/m3e-small",
"m3e-base": "moka-ai/m3e-base",
}
# Embedding model name
EMBEDDING_MODEL = "text2vec"
# Embedding running device
EMBEDDING_DEVICE = "cuda" if torch.cuda.is_available() else "mps" if torch.backends.mps.is_available() else "cpu"
# supported LLM models
# llm_model_dict 處理了loader的一些預設行為，如加載位置，模型名稱，模型處理器實例
# 在以下字典中修改屬性值，以指定本地 LLM 模型存儲位置
# 如將 "chatglm-6b" 的 "local_model_path" 由 None 修改為 "User/Downloads/chatglm-6b"
# 此處請寫絕對路徑
llm_model_dict = {
"chatglm-6b-int4-qe": {
"name": "chatglm-6b-int4-qe",
"pretrained_model_name": "THUDM/chatglm-6b-int4-qe",
"local_model_path": None,
"provides": "ChatGLM"
},
"chatglm-6b-int4": {
"name": "chatglm-6b-int4",
"pretrained_model_name": "THUDM/chatglm-6b-int4",
"local_model_path": None,
"provides": "ChatGLM"
},
"chatglm-6b-int8": {
"name": "chatglm-6b-int8",
"pretrained_model_name": "THUDM/chatglm-6b-int8",
"local_model_path": None,
"provides": "ChatGLM"
},
"chatglm-6b": {
"name": "chatglm-6b",
"pretrained_model_name": "THUDM/chatglm-6b",
"local_model_path": None,
"provides": "ChatGLM"
},
"chatglm2-6b": {
"name": "chatglm2-6b",
"pretrained_model_name": "THUDM/chatglm2-6b",
"local_model_path": None,
"provides": "ChatGLM"
},
"chatglm2-6b-int4": {
"name": "chatglm2-6b-int4",
"pretrained_model_name": "THUDM/chatglm2-6b-int4",
"local_model_path": None,
"provides": "ChatGLM"
},
"chatglm2-6b-int8": {
"name": "chatglm2-6b-int8",
"pretrained_model_name": "THUDM/chatglm2-6b-int8",
"local_model_path": None,
"provides": "ChatGLM"
},
"chatyuan": {
"name": "chatyuan",
"pretrained_model_name": "ClueAI/ChatYuan-large-v2",
"local_model_path": None,
"provides": None
},
"moss": {
"name": "moss",
"pretrained_model_name": "fnlp/moss-moon-003-sft",
"local_model_path": None,
"provides": "MOSSLLM"
},
"vicuna-13b-hf": {
"name": "vicuna-13b-hf",
"pretrained_model_name": "vicuna-13b-hf",
"local_model_path": None,
"provides": "LLamaLLM"
},
# 通過 fastchat 調用的模型請參考如下格式
"fastchat-chatglm-6b": {
"name": "chatglm-6b", # "name"修改為fastchat服務中的"model_name"
"pretrained_model_name": "chatglm-6b",
"local_model_path": None,
"provides": "FastChatOpenAILLM", # 使用fastchat api時，需保證"provides"為"FastChatOpenAILLM"
"api_base_url": "http://localhost:8000/v1" # "name"修改為fastchat服務中的"api_base_url"
},
"fastchat-chatglm2-6b": {
"name": "chatglm2-6b", # "name"修改為fastchat服務中的"model_name"
"pretrained_model_name": "chatglm2-6b",
"local_model_path": None,
"provides": "FastChatOpenAILLM", # 使用fastchat api時，需保證"provides"為"FastChatOpenAILLM"
"api_base_url": "http://localhost:8000/v1" # "name"修改為fastchat服務中的"api_base_url"
},
# 通過 fastchat 調用的模型請參考如下格式
"fastchat-vicuna-13b-hf": {
"name": "vicuna-13b-hf", # "name"修改為fastchat服務中的"model_name"
"pretrained_model_name": "vicuna-13b-hf",
"local_model_path": None,
"provides": "FastChatOpenAILLM", # 使用fastchat api時，需保證"provides"為"FastChatOpenAILLM"
"api_base_url": "http://localhost:8000/v1" # "name"修改為fastchat服務中的"api_base_url"
},
}
# LLM 名稱
LLM_MODEL = "chatglm-6b"
# 量化加載8bit 模型
LOAD_IN_8BIT = False
# Load the model with bfloat16 precision. Requires NVIDIA Ampere GPU.
BF16 = False
# 本地lora存放的位置
LORA_DIR = "loras/"
# LLM lora path，默認為空，如果有請直接指定文件夾路徑
LLM_LORA_PATH = ""
USE_LORA = True if LLM_LORA_PATH else False
# LLM streaming reponse
STREAMING = True
# Use p-tuning-v2 PrefixEncoder
USE_PTUNING_V2 = False
# LLM running device
LLM_DEVICE = "cuda" if torch.cuda.is_available() else "mps" if torch.backends.mps.is_available() else "cpu"
# 知識庫默認存儲路徑
KB_ROOT_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), "knowledge_base")
# 基于上下文的prompt模版，請務必保留"{question}"和"{context}"
PROMPT_TEMPLATE = """已知信息：
{context}
根據上述已知信息，簡潔和專業(yè)的來回答用戶的問題。如果無法從中得到答案，請說 “根據已知信息無法回答該問題” 或 “沒有提供足夠的相關信息”，不允許在答案中添加編造成分，答案請使用中文。 問題是：{question}"""
# 緩存知識庫數量,如果是ChatGLM2,ChatGLM2-int4,ChatGLM2-int8模型若檢索效果不好可以調成’10’
CACHED_VS_NUM = 1
# 文本分句長度
SENTENCE_SIZE = 100
# 匹配后單段上下文長度
CHUNK_SIZE = 250
# 傳入LLM的歷史記錄長度
LLM_HISTORY_LEN = 3
# 知識庫檢索時返回的匹配內容條數
VECTOR_SEARCH_TOP_K = 5
# 知識檢索內容相關度 Score, 數值范圍約為0-1100，如果為0，則不生效，經測試設置為小于500時，匹配結果更精準
VECTOR_SEARCH_SCORE_THRESHOLD = 0
NLTK_DATA_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), "nltk_data")
FLAG_USER_NAME = uuid.uuid4().hex
logger.info(f"""
loading model config
llm device: {LLM_DEVICE}
embedding device: {EMBEDDING_DEVICE}
dir: {os.path.dirname(os.path.dirname(__file__))}
flagging username: {FLAG_USER_NAME}
""")
# 是否開啟跨域，默認為False，如果需要開啟，請設置為True
# is open cross domain
OPEN_CROSS_DOMAIN = False
# Bing 搜索必備變量
# 使用 Bing 搜索需要使用 Bing Subscription Key,需要在azure port中申請試用bing search
# 具體申請方式請見
# https://learn.microsoft.com/en-us/bing/search-apis/bing-web-search/create-bing-search-service-resource
# 使用python創(chuàng)建bing api 搜索實例詳見:
# https://learn.microsoft.com/en-us/bing/search-apis/bing-web-search/quickstarts/rest/python
BING_SEARCH_URL = "https://api.bing.microsoft.com/v7.0/search"
# 注意不是bing Webmaster Tools的api key，
# 此外，如果是在服務器上，報Failed to establish a new connection: [Errno 110] Connection timed out
# 是因為服務器加了防火墻，需要聯系管理員加白名單，如果公司的服務器的話，就別想了GG
BING_SUBSCRIPTION_KEY = ""
# 是否開啟中文標題加強，以及標題增強的相關配置
# 通過增加標題判斷，判斷哪些文本為標題，并在metadata中進行標記；
# 然后將文本與往上一級的標題進行拼合，實現文本信息的增強。
ZH_TITLE_ENHANCE = False3.4?loader：文檔加載與text轉換
3.4.1?loader/pdf_loader.py
# 導入類型提示模塊，用于強化代碼的可讀性和健壯性
from typing import List
# 導入UnstructuredFileLoader，這是一個從非結構化文件中加載文檔的類
from langchain.document_loaders.unstructured import UnstructuredFileLoader
# 導入PaddleOCR，這是一個開源的OCR工具，用于從圖片中識別和讀取文字
from paddleocr import PaddleOCR
# 導入os模塊，用于處理文件和目錄
import os
# 導入fitz模塊，用于處理PDF文件
import fitz
# 導入nltk模塊，用于處理文本數據
import nltk
# 導入模型配置文件中的NLTK_DATA_PATH，這是nltk數據的路徑
from configs.model_config import NLTK_DATA_PATH
# 設置nltk數據的路徑，將模型配置中的路徑添加到nltk的數據路徑中
nltk.data.path = [NLTK_DATA_PATH] + nltk.data.path
# 定義一個類，UnstructuredPaddlePDFLoader，該類繼承自UnstructuredFileLoader
class UnstructuredPaddlePDFLoader(UnstructuredFileLoader):
# 定義一個內部方法_get_elements，返回一個列表
def _get_elements(self) -> List:
# 定義一個內部函數pdf_ocr_txt，用于從pdf中進行OCR并輸出文本文件
def pdf_ocr_txt(filepath, dir_path="tmp_files"):
# 將dir_path與filepath的目錄部分合并成一個新的路徑
full_dir_path = os.path.join(os.path.dirname(filepath), dir_path)
# 如果full_dir_path對應的目錄不存在，則創(chuàng)建這個目錄
if not os.path.exists(full_dir_path):
os.makedirs(full_dir_path)
# 創(chuàng)建一個PaddleOCR實例，設置一些參數
ocr = PaddleOCR(use_angle_cls=True, lang="ch", use_gpu=False, show_log=False)
# 打開pdf文件
doc = fitz.open(filepath)
# 創(chuàng)建一個txt文件的路徑
txt_file_path = os.path.join(full_dir_path, f"{os.path.split(filepath)[-1]}.txt")
# 創(chuàng)建一個臨時的圖片文件路徑
img_name = os.path.join(full_dir_path, 'tmp.png')
# 打開txt_file_path對應的文件，并以寫模式打開
with open(txt_file_path, 'w', encoding='utf-8') as fout:
# 遍歷pdf的所有頁面
for i in range(doc.page_count):
# 獲取當前頁面
page = doc[i]
# 獲取當前頁面的文本內容，并寫入txt文件
text = page.get_text("")
fout.write(text)
fout.write("\n")
# 獲取當前頁面的所有圖片
img_list = page.get_images()
# 遍歷所有圖片
for img in img_list:
# 將圖片轉換為Pixmap對象
pix = fitz.Pixmap(doc, img[0])
# 如果圖片有顏色信息，則將其轉換為RGB格式
if pix.n - pix.alpha >= 4:
pix = fitz.Pixmap(fitz.csRGB, pix)
# 保存圖片
pix.save(img_name)
# 對圖片進行OCR識別
result = ocr.ocr(img_name)
# 從OCR結果中提取文本，并寫入txt文件
ocr_result = [i[1][0] for line in result for i in line]
fout.write("\n".join(ocr_result))
# 如果圖片文件存在，則刪除它
if os.path.exists(img_name):
os.remove(img_name)
# 返回txt文件的路徑
return txt_file_path
# 調用上面定義的函數，獲取txt文件的路徑
txt_file_path = pdf_ocr_txt(self.file_path)
# 導入partition_text函數，該函數用于將文本文件分塊
from unstructured.partition.text import partition_text
# 對txt文件進行分塊，并返回分塊結果
return partition_text(filename=txt_file_path, **self.unstructured_kwargs)
# 運行入口
if __name__ == "__main__":
# 導入sys模塊，用于操作Python的運行環(huán)境
import sys
# 將當前文件的上一級目錄添加到Python的搜索路徑中
sys.path.append(os.path.dirname(os.path.dirname(__file__)))
# 定義一個pdf文件的路徑
filepath = os.path.join(os.path.dirname(os.path.dirname(__file__)), "knowledge_base", "samples", "content", "test.pdf")
# 創(chuàng)建一個UnstructuredPaddlePDFLoader的實例
loader = UnstructuredPaddlePDFLoader(filepath, mode="elements")
# 加載文檔
docs = loader.load()
# 遍歷并打印所有文檔
for doc in docs:
print(doc)// 待更..
3.5?textsplitter：文檔切分
3.5.1?textsplitter/ali_text_splitter.py
ali_text_splitter.py的代碼如下所示
# 導入CharacterTextSplitter模塊，用于文本切分
from langchain.text_splitter import CharacterTextSplitter
import re # 導入正則表達式模塊，用于文本匹配和替換
from typing import List # 導入List類型，用于指定返回的數據類型
# 定義一個新的類AliTextSplitter，繼承自CharacterTextSplitter
class AliTextSplitter(CharacterTextSplitter):
# 類的初始化函數，如果參數pdf為True，那么使用pdf文本切分規(guī)則，否則使用默認規(guī)則
def __init__(self, pdf: bool = False, **kwargs):
# 調用父類的初始化函數，接收傳入的其他參數
super().__init__(**kwargs)
self.pdf = pdf # 將pdf參數保存為類的成員變量
# 定義文本切分方法，輸入參數為一個字符串，返回值為字符串列表
def split_text(self, text: str) -> List[str]:
if self.pdf: # 如果pdf參數為True，那么對文本進行預處理
# 替換掉連續(xù)的3個及以上的換行符為一個換行符
text = re.sub(r"\n{3,}", r"\n", text)
# 將所有的空白字符（包括空格、制表符、換頁符等）替換為一個空格
text = re.sub('\s', " ", text)
# 將連續(xù)的兩個換行符替換為一個空字符
text = re.sub("\n\n", "", text)
# 導入pipeline模塊，用于創(chuàng)建一個處理流程
from modelscope.pipelines import pipeline
# 創(chuàng)建一個document-segmentation任務的處理流程
# 用的模型為damo/nlp_bert_document-segmentation_chinese-base，計算設備為cpu
p = pipeline(
task="document-segmentation",
model='damo/nlp_bert_document-segmentation_chinese-base',
device="cpu")
result = p(documents=text) # 對輸入的文本進行處理，返回處理結果
sent_list = [i for i in result["text"].split("\n\t") if i] # 將處理結果按照換行符和制表符進行切分，得到句子列表
return sent_list # 返回句子列表其中，有三點值得注意下
- 參數use_document_segmentation指定是否用語義切分文檔
此處采取的文檔語義分割模型為達摩院開源的：nlp_bert_document-segmentation_chinese-base? (這是其論文) - 另，如果使用模型進行文檔語義切分，那么需要安裝：
modelscope[nlp]：pip install "modelscope[nlp]" -f https://modelscope.oss-cn-beijing.aliyuncs.com/releases/repo.html - 且考慮到使用了三個模型，可能對于低配置gpu不太友好，因此這里將模型load進cpu計算，有需要的話可以替換device為自己的顯卡id
3.6 knowledge_base：存儲用戶上傳的文件并向量化
knowledge_bas下面有兩個文件，一個content 即用戶上傳的原始文件，vector_store則用于存儲向量庫?件，即本地知識庫本體，因為content因人而異 誰上傳啥就是啥 所以沒啥好分析，而vector_store下面則有兩個文件，一個index.faiss，一個index.pkl
3.7 chains：向量搜索/匹配
如之前所述，本節(jié)開頭圖中“FAISS索引、FAISS搜索”中的“FAISS”是Facebook AI推出的一種用于有效搜索大規(guī)模高維向量空間中相似度的庫，在大規(guī)模數據集中快速找到與給定向量最相似的向量是很多AI應用的重要組成部分，例如在推薦系統(tǒng)、自然語言處理、圖像檢索等領域
3.7.1 chains/modules /vectorstores.py文件：根據查詢向量query在向量數據庫中查找與query相似的文本向量
主要是關于
- FAISS (Facebook AI Similarity Search)的使用，具體體現在max_marginal_relevance_search_by_vector 中(如下圖的最上面部分)
- 以及一個FAISS向量存儲類(FAISSVS，FAISSVS類繼承自FAISS類)的定義，包含兩個方法
一個 max_marginal_relevance_search (如下圖的中間部分，其最后會調用上面的max_marginal_relevance_search_by_vector)
一個 __from (如下圖的最下面部分)
接下來，我們逐一分析下這幾個函數
-
max_marginal_relevance_search
分兩步，給定查詢語句，首先將查詢語句轉換為嵌入向量「embedding = self.embedding_function(query)」，然后調用?max_marginal_relevance_search_by_vector?函數進行MMR搜索下面看下其中?max_marginal_relevance_search_by_vector?的實現# 使用最大邊際相關性返回被選中的文本 def max_marginal_relevance_search( self, query: str, # 查詢 k: int = 4, # 返回的文檔數量，默認為 4 fetch_k: int = 20, # 用于傳遞給 MMR 算法的抓取文檔數量 **kwargs: Any, ) -> List[Tuple[Document, float]]: # 查詢向量化 embedding = self.embedding_function(query) # 調用：max_marginal_relevance_search_by_vector docs = self.max_marginal_relevance_search_by_vector(embedding, k, fetch_k) return docs
該函數通過給定的嵌入向量，使用最大邊際相關性(Maximal Marginal Relevance, MMR)方法來返回相關的文本
MMR是一種解決查詢結果多樣性和相關性的算法，具體來說，它不僅要求返回的文本與查詢盡可能相似，而且希望返回的文本集之間盡可能多樣注意，上面第6-7行代碼中，直接調用的index的search函數# 使用最大邊際相關性返回被選中的文檔，最大邊際相關性旨在優(yōu)化查詢的相似性和選定文本之間的多樣性 def max_marginal_relevance_search_by_vector( self, embedding: List[float], k: int = 4, fetch_k: int = 20, **kwargs: Any ) -> List[Tuple[Document, float]]: # 使用索引在文本中搜索與嵌入向量相似的內容，返回最相似的fetch_k個文本的得分和索引 scores, indices = self.index.search(np.array([embedding], dtype=np.float32), fetch_k) # 通過索引從文本中重構出嵌入向量，-1表示沒有足夠的文本返回 embeddings = [self.index.reconstruct(int(i)) for i in indices[0] if i != -1] # 使用最大邊際相關性算法選擇出k個最相關的文本 mmr_selected = maximal_marginal_relevance( np.array([embedding], dtype=np.float32), embeddings, k=k ) selected_indices = [indices[0][i] for i in mmr_selected] # 獲取被選中的文本的索引 selected_scores = [scores[0][i] for i in mmr_selected] # 獲取被選中的文本的得分 docs = [] for i, score in zip(selected_indices, selected_scores): # 對于每個被選中的文本索引和得分 if i == -1: # 如果索引為-1，表示沒有足夠的文本返回 continue _id = self.index_to_docstore_id[i] # 通過索引獲取文本的id doc = self.docstore.search(_id) # 通過id在文檔庫中搜索文本 if not isinstance(doc, Document): # 如果搜索到的文本不是Document類型，拋出錯誤 raise ValueError(f"Could not find document for id {_id}, got {doc}") docs.append((doc, score)) # 將文本和得分添加到結果列表中 return docs # 返回結果列表
? ? # 使用索引在文本中搜索與嵌入向量相似的內容，返回最相似的fetch_k個文本的得分和索引
? ? scores, indices = self.index.search(np.array([embedding], dtype=np.float32), fetch_k)
這里面就有來頭了，通過和我司杜老師的討論確定，這個search是根據構建索引時所用的度量指標找到最近的k個向量的，在構建索引時
? 如果是faiss.IndexFlatIP，就是內積(METRIC_INNER_PRODUCT)，可以認為是余弦相似度
? 如果是faiss.IndexFlatL2，就是歐氏距離(METRIC_L2)，更多計算距離的方式詳見上文的2.2.2節(jié)
其具體的代碼實現如下所示(來源：faiss/IndexFlat.cpp#L27)void IndexFlat::search( idx_t n, // 搜索的查詢向量的數量 const float* x, // 指向查詢向量數據的指針 idx_t k, // 每個查詢向量返回的最近鄰個數 float* distances, // 返回的距離數組 idx_t* labels, // 返回的標簽數組 const SearchParameters* params) const { // 搜索參數 // 如果params非空，則使用params中的選擇器，否則使用nullptr IDSelector* sel = params ? params->sel : nullptr; // 檢查k（最近鄰的數量）必須大于0 FAISS_THROW_IF_NOT(k > 0); // distances和labels被視為堆（用于存儲最近鄰的結果） if (metric_type == METRIC_INNER_PRODUCT) { // 如果度量類型是內積 float_minheap_array_t res = {size_t(n), size_t(k), labels, distances}; // 使用內積計算最近鄰 knn_inner_product(x, get_xb(), d, n, ntotal, &res, sel); } else if (metric_type == METRIC_L2) { // 如果度量類型是L2距離 float_maxheap_array_t res = {size_t(n), size_t(k), labels, distances}; // 使用L2距離計算最近鄰 knn_L2sqr(x, get_xb(), d, n, ntotal, &res, nullptr, sel); } else if (is_similarity_metric(metric_type)) { // 如果度量類型是其他相似度度量 float_minheap_array_t res = {size_t(n), size_t(k), labels, distances}; // 使用其他相似度度量計算最近鄰 knn_extra_metrics( x, get_xb(), d, n, ntotal, metric_type, metric_arg, &res); } else { // 其他情況 FAISS_THROW_IF_NOT(!sel); // 確保選擇器為空 float_maxheap_array_t res = {size_t(n), size_t(k), labels, distances}; // 使用其他相似度度量計算最近鄰 knn_extra_metrics( x, get_xb(), d, n, ntotal, metric_type, metric_arg, &res); } } -
__from
用于從一組文本和對應的嵌入向量創(chuàng)建一個FAISSVS實例。該方法首先創(chuàng)建一個FAISS索引并添加嵌入向量，然后創(chuàng)建一個文本存儲以存儲與每個嵌入向量關聯的文本從上面代碼的第11-13行中# 從給定的文本、嵌入向量、元數據等信息構建一個FAISS索引對象 def __from( cls, texts: List[str], # 文本列表，每個文本將被轉化為一個文本對象 embeddings: List[List[float]], # 對應文本的嵌入向量列表 embedding: Embeddings, # 嵌入向量生成器，用于將查詢語句轉化為嵌入向量 metadatas: Optional[List[dict]] = None, **kwargs: Any, ) -> FAISS: faiss = dependable_faiss_import() # 導入FAISS庫 index = faiss.IndexFlatIP(len(embeddings[0])) # 使用FAISS庫創(chuàng)建一個新的索引，索引的維度等于嵌入文本向量的長度 index.add(np.array(embeddings, dtype=np.float32)) # 將嵌入向量添加到FAISS索引中 # quantizer = faiss.IndexFlatL2(len(embeddings[0])) # index = faiss.IndexIVFFlat(quantizer, len(embeddings[0]), 100) # index.train(np.array(embeddings, dtype=np.float32)) # index.add(np.array(embeddings, dtype=np.float32)) documents = [] for i, text in enumerate(texts): # 對于每一段文本 # 獲取對應的元數據，如果沒有提供元數據則使用空字典 metadata = metadatas[i] if metadatas else {} # 創(chuàng)建一個文本對象并添加到文本列表中 documents.append(Document(page_content=text, metadata=metadata)) # 為每個文本生成一個唯一的ID index_to_id = {i: str(uuid.uuid4()) for i in range(len(documents))} # 創(chuàng)建一個文本庫，用于存儲文本對象和對應的ID docstore = InMemoryDocstore( {index_to_id[i]: doc for i, doc in enumerate(documents)} ) # 返回FAISS對象 return cls(embedding.embed_query, index, docstore, index_to_id)
?faiss = dependable_faiss_import() ? ? ?# 導入FAISS庫
? ? index = faiss.IndexFlatIP(len(embeddings[0])) ? ? ?# 使用FAISS庫創(chuàng)建一個新的索引，索引的維度等于嵌入文本向量的長度
? ? index.add(np.array(embeddings, dtype=np.float32)) ?# 將嵌入向量添加到FAISS索引中
可知，構建index的時候就已經指定了用IndexFlatIP(余弦相似度)的方式計算距離(你可以通過這個代碼鏈接驗證下：chains/modules/vectorstores.py#L103)
以上就是這段代碼的主要內容，通過使用FAISS和MMR，它可以幫助我們在大量文本中找到與給定查詢最相關的文本
3.7.2 chains /local_doc_qa.py代碼文件：向量搜索
- 導入包和模塊
代碼開始的部分是一系列的導入語句，導入了必要的 Python 包和模塊，包括文件加載器，文本分割器，模型配置，以及一些 Python 內建模塊和其他第三方庫 - 改寫 HuggingFaceEmbeddings 類的哈希方法
代碼定義了一個名為 _embeddings_hash 的函數，并將其賦值給 HuggingFaceEmbeddings 類的 __hash__ 方法。這樣做的目的是使 HuggingFaceEmbeddings 對象可以被哈希，即可以作為字典的鍵或者被加入到集合中 - 載入向量存儲器
定義了一個名為 load_vector_store 的函數，這個函數用于從本地加載一個向量存儲器，返回 FAISS 類的對象。其中使用了 lru_cache 裝飾器，可以緩存最近使用的 CACHED_VS_NUM 個結果，提高代碼效率 - 文件樹遍歷
tree 函數是一個遞歸函數，用于遍歷指定目錄下的所有文件，返回一個包含所有文件的完整路徑和文件名的列表。它可以忽略指定的文件或目錄 - 加載文件：
load_file 函數根據文件后綴名選擇合適的加載器和文本分割器，加載并分割文件 - 生成提醒:
generate_prompt 函數用于根據相關文檔和查詢生成一個提醒。提醒的模板由 prompt_template 參數提供 - 創(chuàng)建文檔列表
search_result2docs# 創(chuàng)建一個空列表，用于存儲文檔 def search_result2docs(search_results): docs = [] # 對于搜索結果中的每一項 for result in search_results: # 創(chuàng)建一個文檔對象 # 如果結果中包含"snippet"關鍵字，則其值作為頁面內容，否則頁面內容為空字符串 # 如果結果中包含"link"關鍵字，則其值作為元數據中的源鏈接，否則源鏈接為空字符串 # 如果結果中包含"title"關鍵字，則其值作為元數據中的文件名，否則文件名為空字符串 doc = Document(page_content=result["snippet"] if "snippet" in result.keys() else "", metadata={"source": result["link"] if "link" in result.keys() else "", "filename": result["title"] if "title" in result.keys() else ""}) # 將創(chuàng)建的文檔對象添加到列表中 docs.append(doc) # 返回文檔列表 return docs
之后，定義了一個名為 LocalDocQA 的類，主要用于基于文檔的問答任務?；谖臋n的問答任務的主要功能是，根據一組給定的文檔(這里被稱為知識庫)以及用戶輸入的問題，返回一個答案，LocalDocQA 類的主要方法包括：
- init_cfg()：此方法初始化一些變量，包括將 llm_model（一個語言模型用于生成答案）分配給 self.llm，將一個基于HuggingFace的嵌入模型分配給 self.embeddings，將輸入參數 top_k 分配給 self.top_k
- init_knowledge_vector_store()：此方法負責初始化知識向量庫。它首先檢查輸入的文件路徑，對于路徑中的每個文件，將文件內容加載到 Document 對象中，然后將這些文檔轉換為嵌入向量，并將它們存儲在向量庫中
- one_knowledge_add()：此方法用于向知識庫中添加一個新的知識文檔。它將輸入的標題和內容創(chuàng)建為一個 Document 對象，然后將其轉換為嵌入向量，并添加到向量庫中
-
get_knowledge_based_answer()：此方法是基于給定的知識庫和用戶輸入的問題，來生成一個答案。它首先根據用戶輸入的問題找到知識庫中最相關的文檔，然后生成一個包含相關文檔和用戶問題的提示，將提示傳遞給 llm_model 來生成答案
且注意一點，這個函數調用了上面已經實現好的：similarity_search_with_score -
get_knowledge_based_conent_test()：此方法是為了測試的，它將返回與輸入查詢最相關的文檔和查詢提示
? ? # query ? ? ?查詢內容
? ? # vs_path ? ?知識庫路徑
? ? # chunk_conent ? 是否啟用上下文關聯
? ? # score_threshold ? ?搜索匹配score閾值
? ? # vector_search_top_k ? 搜索知識庫內容條數，默認搜索5條結果
? ? # chunk_sizes ? ?匹配單段內容的連接上下文長度
? ? def get_knowledge_based_conent_test(self, query, vs_path, chunk_conent,
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? score_threshold=VECTOR_SEARCH_SCORE_THRESHOLD,
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? vector_search_top_k=VECTOR_SEARCH_TOP_K, chunk_size=CHUNK_SIZE): -
get_search_result_based_answer()：此方法與 get_knowledge_based_answer() 類似，不過這里使用的是 bing_search 的結果作為知識庫
如你所見，這個函數和上面那個函數的主要區(qū)別在于，這個函數是直接利用搜索引擎的搜索結果來生成回答的，而上面那個函數是通過查詢相似度搜索來找到最相關的文本，然后基于這些文本生成回答的
def get_search_result_based_answer(self, query, chat_history=[], streaming: bool = STREAMING): # 對查詢進行 Bing 搜索，并獲取搜索結果 results = bing_search(query) # 將搜索結果轉化為文本的形式 result_docs = search_result2docs(results) # 生成用于提問的提示語 prompt = generate_prompt(result_docs, query) # 通過 LLM（長語言模型）生成回答 for answer_result in self.llm.generatorAnswer(prompt=prompt, history=chat_history, streaming=streaming): # 獲取回答的文本 resp = answer_result.llm_output["answer"] # 獲取聊天歷史 history = answer_result.history # 將聊天歷史中的最后一項的提問替換為當前的查詢 history[-1][0] = query # 組裝回答的結果 response = {"query": query, "result": resp, "source_documents": result_docs} # 返回回答的結果和聊天歷史 yield response, history
而這個bing_search則在3.1.2節(jié)中已經定義 - 接下來是分別用于從向量存儲中刪除文件、更新文件以及列出文件的三個方法
delete_file_from_vector_store
update_file_from_vector_store
list_file_from_vector_store# 刪除向量存儲中的文件 def delete_file_from_vector_store(self, filepath: str or List[str], # 文件路徑，可以是單個文件或多個文件列表 vs_path): # 向量存儲路徑 vector_store = load_vector_store(vs_path, self.embeddings) # 從給定路徑加載向量存儲 status = vector_store.delete_doc(filepath) # 刪除指定文件 return status # 返回刪除狀態(tài) # 更新向量存儲中的文件 def update_file_from_vector_store(self, filepath: str or List[str], # 需要更新的文件路徑，可以是單個文件或多個文件列表 vs_path, # 向量存儲路徑 docs: List[Document],): # 需要更新的文件內容，文件以文檔形式給出 vector_store = load_vector_store(vs_path, self.embeddings) # 從給定路徑加載向量存儲 status = vector_store.update_doc(filepath, docs) # 更新指定文件 return status # 返回更新狀態(tài) # 列出向量存儲中的文件 def list_file_from_vector_store(self, vs_path, # 向量存儲路徑 fullpath=False): # 是否返回完整路徑，如果為 False，則只返回文件名 vector_store = load_vector_store(vs_path, self.embeddings) # 從給定路徑加載向量存儲 docs = vector_store.list_docs() # 列出所有文件 if fullpath: # 如果需要完整路徑 return docs # 返回完整路徑列表 else: # 如果只需要文件名 return [os.path.split(doc)[-1] for doc in docs] # 用 os.path.split 將路徑和文件名分離，只返回文件名列表
__main__部分的代碼是 LocalDocQA 類的實例化和使用示例
- 它首先初始化了一個 llm_model_ins 對象
- 然后創(chuàng)建了一個 LocalDocQA 的實例并調用其 init_cfg() 方法進行初始化
- 之后，它指定了一個查詢和知識庫的路徑
- 然后調用 get_knowledge_based_answer() 或 get_search_result_based_answer() 方法獲取基于該查詢的答案，并打印出答案和來源文檔的信息
3.7.3 chains/text_load.py
chain這個文件夾下 還有最后一個項目文件(langchain-ChatGLM/text_load.py at master · imClumsyPanda/langchain-ChatGLM · GitHub)，如下所示
import os
import pinecone
from tqdm import tqdm
from langchain.llms import OpenAI
from langchain.text_splitter import SpacyTextSplitter
from langchain.document_loaders import TextLoader
from langchain.document_loaders import DirectoryLoader
from langchain.indexes import VectorstoreIndexCreator
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.vectorstores import Pinecone
#一些配置文件
openai_key="你的key" # 注冊 openai.com 后獲得
pinecone_key="你的key" # 注冊 app.pinecone.io 后獲得
pinecone_index="你的庫" #app.pinecone.io 獲得
pinecone_environment="你的Environment" # 登錄pinecone后，在indexes頁面 查看Environment
pinecone_namespace="你的Namespace" #如果不存在自動創(chuàng)建
#科學上網你懂得
os.environ['HTTP_PROXY'] = 'http://127.0.0.1:7890'
os.environ['HTTPS_PROXY'] = 'http://127.0.0.1:7890'
#初始化pinecone
pinecone.init(
api_key=pinecone_key,
environment=pinecone_environment
)
index = pinecone.Index(pinecone_index)
#初始化OpenAI的embeddings
embeddings = OpenAIEmbeddings(openai_api_key=openai_key)
#初始化text_splitter
text_splitter = SpacyTextSplitter(pipeline='zh_core_web_sm',chunk_size=1000,chunk_overlap=200)
# 讀取目錄下所有后綴是txt的文件
loader = DirectoryLoader('../docs', glob="**/*.txt", loader_cls=TextLoader)
#讀取文本文件
documents = loader.load()
# 使用text_splitter對文檔進行分割
split_text = text_splitter.split_documents(documents)
try:
for document in tqdm(split_text):
# 獲取向量并儲存到pinecone
Pinecone.from_documents([document], embeddings, index_name=pinecone_index)
except Exception as e:
print(f"Error: {e}")
quit()3.8 vectorstores：MyFAISS.py
兩個文件，一個__init__.py (就一行代碼：from .MyFAISS import MyFAISS)，另一個MyFAISS.py，如下代碼所示
# 從langchain.vectorstores庫導入FAISS
from langchain.vectorstores import FAISS
# 從langchain.vectorstores.base庫導入VectorStore
from langchain.vectorstores.base import VectorStore
# 從langchain.vectorstores.faiss庫導入dependable_faiss_import
from langchain.vectorstores.faiss import dependable_faiss_import
from typing import Any, Callable, List, Dict # 導入類型檢查庫
from langchain.docstore.base import Docstore # 從langchain.docstore.base庫導入Docstore
# 從langchain.docstore.document庫導入Document
from langchain.docstore.document import Document
import numpy as np # 導入numpy庫，用于科學計算
import copy # 導入copy庫，用于數據復制
import os # 導入os庫，用于操作系統(tǒng)相關的操作
from configs.model_config import * # 從configs.model_config庫導入所有內容
# 定義MyFAISS類，繼承自FAISS和VectorStore兩個父類
class MyFAISS(FAISS, VectorStore):接下來，逐一實現以下函數
3.8.1?定義類的初始化函數：__init__
# 定義類的初始化函數
def __init__(
self,
embedding_function: Callable,
index: Any,
docstore: Docstore,
index_to_docstore_id: Dict[int, str],
normalize_L2: bool = False,
):
# 調用父類FAISS的初始化函數
super().__init__(embedding_function=embedding_function,
index=index,
docstore=docstore,
index_to_docstore_id=index_to_docstore_id,
normalize_L2=normalize_L2)
# 初始化分數閾值
self.score_threshold=VECTOR_SEARCH_SCORE_THRESHOLD
# 初始化塊大小
self.chunk_size = CHUNK_SIZE
# 初始化塊內容
self.chunk_conent = False3.8.2?seperate_list：將一個列表分解成多個子列表
# 定義函數seperate_list，將一個列表分解成多個子列表，每個子列表中的元素在原列表中是連續(xù)的
def seperate_list(self, ls: List[int]) -> List[List[int]]:
# TODO: 增加是否屬于同一文檔的判斷
lists = []
ls1 = [ls[0]]
for i in range(1, len(ls)):
if ls[i - 1] + 1 == ls[i]:
ls1.append(ls[i])
else:
lists.append(ls1)
ls1 = [ls[i]]
lists.append(ls1)
return lists3.8.3?similarity_search_with_score_by_vector，根據輸入的向量，查找最接近的k個文本
similarity_search_with_score_by_vector 函數用于通過向量進行相似度搜索，返回與給定嵌入向量最相似的文本和對應的分數
不過，這個函數考慮的細節(jié)比較多，所以代碼長度比較長，為方便大家更好的理解，我把這個函數拆分成5段逐一解釋說明
- 定義相似度搜索的函數，其接受一個向量embedding和一個整數k作為參數，之后導入faiss庫，從而調用faiss庫的search函數(對該search函數的細致分析及具體代碼層面的實現，詳見上文的3.7.1節(jié))，查找與輸入向量最接近的k個向量，并返回他們的分數和索引
并初始化文檔列表docs、文檔ID：id_set，方便后面索引找ID、ID找文本# 定義函數similarity_search_with_score_by_vector，根據輸入的向量，查找最接近的k個文本 def similarity_search_with_score_by_vector( self, embedding: List[float], k: int = 4 ) -> List[Document]: # 調用dependable_faiss_import函數，導入faiss庫 faiss = dependable_faiss_import() # 將輸入的列表轉換為numpy數組，并設置數據類型為float32 vector = np.array([embedding], dtype=np.float32) # 如果需要進行L2歸一化，則調用faiss.normalize_L2函數進行歸一化 if self._normalize_L2: faiss.normalize_L2(vector) # 調用faiss庫的search函數，查找與輸入向量最接近的k個向量，并返回他們的分數和索引 scores, indices = self.index.search(vector, k) # 初始化一個空列表，用于存儲找到的文本 docs = [] # 初始化一個空集合，用于存儲文本的id id_set = set() # 獲取文本庫中文本的數量 store_len = len(self.index_to_docstore_id) # 初始化一個布爾變量，表示是否需要重新排列id列表 rearrange_id_list = False - 通過遍歷上面得到的索引indices和分數scores，以驗證是否滿足要求
即，索引和分數需要滿足一定的條件：比如索引不為-1、分數得大于給定的閾值等
如果滿足，則獲取索引 i 對應的文本ID：_id = self.index_to_docstore_id[i]
最終從文本庫中找到對應ID的文本(相當于索引到ID、ID到文本)：doc = self.docstore.search(_id)# 遍歷找到的索引和分數 for j, i in enumerate(indices[0]): # 如果索引為-1，或者分數小于閾值，則跳過這個索引 if i == -1 or 0 < self.score_threshold < scores[0][j]: # This happens when not enough docs are returned. continue # 如果索引存在于index_to_docstore_id字典中，則獲取對應的文本id if i in self.index_to_docstore_id: _id = self.index_to_docstore_id[i] # 如果索引不存在于index_to_docstore_id字典中，則跳過這個索引 else: continue # 從文本庫中搜索對應id的文本 doc = self.docstore.search(_id) - 對上面得到的文檔做進一步的處理，比如檢查是否為Document類型，如果是，則
? ? ? ? ? ? ? ? # 將文本添加到docs列表中
? ? ? ? ? ? ? ? docs.append(doc)
? ? ? ? ? ? ? ? continue? ? ? ? ? ? # 將文本id添加到id_set集合中，相當于更新文本id列表了
再之后，根據doc.metadata決定是否需要擴展搜索到的文本內容：根據context_expand_method的取值決定是向前擴展，還是向后擴展，或向前向后同時擴展(這樣的操作通常用于搜索或檢索任務中，當希望除了當前的匹配結果外，還獲取其前后的一些相關內容)
? ? ? ? ? ? id_set.add(i)# 如果不需要拆分塊內容，或者文檔的元數據中沒有context_expand字段，或者context_expand字段的值為false，則執(zhí)行以下代碼 if (not self.chunk_conent) or ("context_expand" in doc.metadata and not doc.metadata["context_expand"]): # 匹配出的文本如果不需要擴展上下文則執(zhí)行如下代碼 # 如果搜索到的文本不是Document類型，則拋出異常 if not isinstance(doc, Document): raise ValueError(f"Could not find document for id {_id}, got {doc}") # 在文本的元數據中添加score字段，其值為找到的分數 doc.metadata["score"] = int(scores[0][j]) # 將文本添加到docs列表中 docs.append(doc) continue # 將文本id添加到id_set集合中 id_set.add(i) # 獲取文本的長度 docs_len = len(doc.page_content) # 遍歷范圍在1到i和store_len - i之間的數字k for k in range(1, max(i, store_len - i)): # 初始化一個布爾變量，表示是否需要跳出循環(huán) break_flag = False # 如果文本的元數據中有context_expand_method字段，并且其值為"forward"，則擴展范圍設置為[i + k] if "context_expand_method" in doc.metadata and doc.metadata["context_expand_method"] == "forward": expand_range = [i + k] # 如果文本的元數據中有context_expand_method字段，并且其值為"backward"，則擴展范圍設置為[i - k] elif "context_expand_method" in doc.metadata and doc.metadata["context_expand_method"] == "backward": expand_range = [i - k] # 如果文本的元數據中沒有context_expand_method字段，或者context_expand_method字段的值不是"forward"也不是"backward"，則擴展范圍設置為[i + k, i - k] else: expand_range = [i + k, i - k] - 對擴展內容的處理
對于需要擴展其內容的文檔，代碼將其內容與前后的文檔內容合并，直到滿足一定的條件（例如，總長度最好不要超過chunk_size，或者合并的文檔來源盡可能相同）# 遍歷擴展范圍 for l in expand_range: # 如果l不在id_set集合中，并且l在0到len(self.index_to_docstore_id)之間，則執(zhí)行以下代碼 if l not in id_set and 0 <= l < len(self.index_to_docstore_id): # 獲取l對應的文本id _id0 = self.index_to_docstore_id[l] # 從文本庫中搜索對應id的文本 doc0 = self.docstore.search(_id0) # 如果文本長度加上新文檔的長度大于塊大小，或者新文本的源不等于當前文本的源，則設置break_flag為true，跳出循環(huán) if docs_len + len(doc0.page_content) > self.chunk_size or doc0.metadata["source"] != \ doc.metadata["source"]: break_flag = True break # 如果新文本的源等于當前文本的源，則將新文本的長度添加到文本長度上，將l添加到id_set集合中，設置rearrange_id_list為true elif doc0.metadata["source"] == doc.metadata["source"]: docs_len += len(doc0.page_content) id_set.add(l) rearrange_id_list = True # 如果break_flag為true，則跳出循環(huán) if break_flag: break - 如果對文檔內容進行了擴展，則需要重新整理ID列表
然后對擴展內容文本的ID到index_to_docstore_id中查詢
? 如果在，則從文檔庫中搜索到對應ID的文本，然后深度拷貝出來
? 否則，則從文本庫中搜索對應id的文檔，將新文本的內容添加到當前文本的內容后面# 如果不需要拆分塊內容，或者不需要重新排列id列表，則返回docs列表 if (not self.chunk_conent) or (not rearrange_id_list): return docs # 如果id_set集合的長度為0，并且分數閾值大于0，則返回空列表 if len(id_set) == 0 and self.score_threshold > 0: return [] # 對id_set集合中的元素進行排序，并轉換為列表 id_list = sorted(list(id_set)) # 調用seperate_list函數，將id_list分解成多個子列表 id_lists = self.seperate_list(id_list) # 遍歷id_lists中的每一個id序列 for id_seq in id_lists: # 遍歷id序列中的每一個id for id in id_seq: # 如果id等于id序列的第一個元素，則從文檔庫中搜索對應id的文本，并深度拷貝這個文本 if id == id_seq[0]: _id = self.index_to_docstore_id[id] # doc = self.docstore.search(_id) doc = copy.deepcopy(self.docstore.search(_id)) # 如果id不等于id序列的第一個元素，則從文本庫中搜索對應id的文檔，將新文本的內容添加到當前文本的內容后面 else: _id0 = self.index_to_docstore_id[id] doc0 = self.docstore.search(_id0) doc.page_content += " " + doc0.page_content # 如果搜索到的文本不是Document類型，則拋出異常 if not isinstance(doc, Document): raise ValueError(f"Could not find document for id {_id}, got {doc}") # 計算文本的分數，分數等于id序列中的每一個id在分數列表中對應的分數的最小值 doc_score = min([scores[0][id] for id in [indices[0].tolist().index(i) for i in id_seq if i in indices[0]]]) # 在文本的元數據中添加score字段，其值為文檔的分數 doc.metadata["score"] = int(doc_score) # 將文本添加到docs列表中 docs.append(doc) # 返回docs列表 return docs
3.8.4 delete_doc方法：刪除文本庫中指定來源的文本
#定義了一個名為 delete_doc 的方法，這個方法用于刪除文本庫中指定來源的文本
def delete_doc(self, source: str or List[str]):
# 使用 try-except 結構捕獲可能出現的異常
try:
# 如果 source 是字符串類型
if isinstance(source, str):
# 找出文本庫中所有來源等于 source 的文本的id
ids = [k for k, v in self.docstore._dict.items() if v.metadata["source"] == source]
# 獲取向量存儲的路徑
vs_path = os.path.join(os.path.split(os.path.split(source)[0])[0], "vector_store")
# 如果 source 是列表類型
else:
# 找出文本庫中所有來源在 source 列表中的文本的id
ids = [k for k, v in self.docstore._dict.items() if v.metadata["source"] in source]
# 獲取向量存儲的路徑
vs_path = os.path.join(os.path.split(os.path.split(source[0])[0])[0], "vector_store")
# 如果沒有找到要刪除的文本，返回失敗信息
if len(ids) == 0:
return f"docs delete fail"
# 如果找到了要刪除的文本
else:
# 遍歷所有要刪除的文本id
for id in ids:
# 獲取該id在索引中的位置
index = list(self.index_to_docstore_id.keys())[list(self.index_to_docstore_id.values()).index(id)]
# 從索引中刪除該id
self.index_to_docstore_id.pop(index)
# 從文本庫中刪除該id對應的文本
self.docstore._dict.pop(id)
# TODO: 從 self.index 中刪除對應id，這是一個未完成的任務
# self.index.reset()
# 保存當前狀態(tài)到本地
self.save_local(vs_path)
# 返回刪除成功的信息
return f"docs delete success"
# 捕獲異常
except Exception as e:
# 打印異常信息
print(e)
# 返回刪除失敗的信息
return f"docs delete fail"3.8.5 update_doc和lists_doc
# 定義了一個名為 update_doc 的方法，這個方法用于更新文檔庫中的文檔
def update_doc(self, source, new_docs):
# 使用 try-except 結構捕獲可能出現的異常
try:
# 刪除舊的文檔
delete_len = self.delete_doc(source)
# 添加新的文檔
ls = self.add_documents(new_docs)
# 返回更新成功的信息
return f"docs update success"
# 捕獲異常
except Exception as e:
# 打印異常信息
print(e)
# 返回更新失敗的信息
return f"docs update fail"
# 定義了一個名為 list_docs 的方法，這個方法用于列出文檔庫中所有文檔的來源
def list_docs(self):
# 遍歷文檔庫中的所有文檔，取出每個文檔的來源，轉換為集合，再轉換為列表，最后返回這個列表
return list(set(v.metadata["source"] for v in self.docstore._dict.values()))第四部分 23年9月升級版Langchain-Chatchat的源碼解析
23年9月，原項目LangChain + ChatGLM-6B做了升級，變成如今的Langchain-Chatchat項目
?其主要更新體現在增加了一個sever的文件夾，該文件夾包括
-
chat，包含
__init__.py
chat.py
knowledge_base_chat.py
openai_chat.py
search_engine_chat.py
utils.py -
db，包含
models
? ? __init__.py
? ? base.py
? ? knowledge_base_model.py (即KnowledgeBaseModel的實現，下文4.2.1節(jié)闡述)
? ? knowledge_file_model.py (即KnowledgeFile的實現，下文4.2.2節(jié)闡述)
repository
? ? __init__.py
? ? knowledge_base_repository.py (涉及add_kb_to_db的實現，下文4.3.1節(jié)闡述)
? ? knowledge_file_repository.py (涉及add_flie_to_db/add_docs_to_db的實現，下文4.3.2節(jié)闡述) -
knowledge_base，包含
kb_cache
? ? base.py (涉及KBServiceFactory的實現，下文4.1.2節(jié)闡述)
? ? faiss_cache.py
kb_service
? ? __init__.py
? ? base.py
? ? default_kb_service.py
? ? faiss_kb_service.py
? ? milvus_kb_service.py
? ? pg_kb_service.py
__init__.py
kb_api.py
kb_doc_api.py (這個實現了多文檔問答的最核心邏輯，下文4.1.1節(jié)闡述)
migrate.py
utils.py - model_workers
- static
等分文件夾
4.1 server/knowledge_base：基于批量文檔的企業(yè)知識庫問答
該項目的最新版中實現了基于批量文檔的問答，比如
# 開始遍歷自定義的文檔集合（docs）
for file_name, v in docs.items():
try:
# 對于v中的每個條目，檢查它是否已經是Document類型
# 如果不是，那么將其轉換為Document對象
v = [x if isinstance(x, Document) else Document(**x) for x in v]
# 根據文件名和知識庫名稱創(chuàng)建KnowledgeFile對象
kb_file = KnowledgeFile(filename=file_name, knowledge_base_name=knowledge_base_name)
# 在知識庫中更新該文件的文檔
kb.update_doc(kb_file, docs=v, not_refresh_vs_cache=True)
# ...4.1.1?knowledge_base /kb_doc_api.py
以下是對該項目文件的逐行分析：Langchain-Chatchat/server/knowledge_base /kb_doc_api.py
-
導入模塊:
- 導入了一些標準庫，如
os和urllib，分別用于操作系統(tǒng)操作和URL解析 - 導入
fastapi相關的模塊，這是一個現代、高速的 web 框架，用于構建 API - 導入自定義的模塊和配置，如模型配置和知識庫的工具函數
- 導入了一些標準庫，如
-
DocumentWithScore類:
- 這是一個繼承自
Document類的子類，增加了一個score字段。這很可能是在文檔搜索時返回搜索相關度得分的一個數據結構
- 這是一個繼承自
-
search_docs函數:
- 這個函數用于搜索知識庫中的文檔。
- 它接受查詢字符串、知識庫名稱、最大返回文檔數量和評分閾值作為參數。
- 使用
KBServiceFactory來獲取對應的知識庫服務，然后在該知識庫中搜索文檔。 - 返回搜索到的與查詢相關的文檔列表，每個文檔都帶有一個相關度得分
# 定義一個用于搜索文檔的函數 def search_docs(query: str = Body(..., description="用戶輸入", examples=["你好"]), knowledge_base_name: str = Body(..., description="知識庫名稱", examples=["samples"]), top_k: int = Body(VECTOR_SEARCH_TOP_K, description="匹配向量數"), score_threshold: float = Body(SCORE_THRESHOLD, description="知識庫匹配相關度閾值，取值范圍在0-1之間，SCORE越小，相關度越高，取到1相當于不篩選，建議設置在0.5左右", ge=0, le=1), ) -> List[DocumentWithScore]: # 根據知識庫名稱獲取相應的知識庫服務實例 kb = KBServiceFactory.get_service_by_name(knowledge_base_name) # 如果沒有找到對應的知識庫服務實例，返回空列表 if kb is None: return [] # 調用知識庫服務的search_docs方法來搜索與查詢字符串匹配的文檔 docs = kb.search_docs(query, top_k, score_threshold) # 將搜索到的文檔轉換為DocumentWithScore對象，包括文檔內容和匹配分數 data = [DocumentWithScore(**x[0].dict(), score=x[1]) for x in docs] # 返回帶分數的匹配文檔列表 return data總之，這個search_docs函數是為了在給定的知識庫中搜索與查詢字符串匹配的文檔，并返回最相關的top_k個文檔及其匹配分數
-
list_files函數
- 用于列出指定知識庫中的所有文件。
- 先進行知識庫名稱的驗證，然后使用
KBServiceFactory來獲取對應的知識庫服務。 - 如果知識庫存在，它將返回知識庫中所有的文檔名稱。
-
_save_files_in_thread函數
- 這是一個私有函數，用于在后臺線程中保存上傳的文件到指定的知識庫。
- 它的內部定義了一個
save_file函數，用于保存單個文件。 - 如果文件已存在，并且用戶沒有選擇覆蓋，它將檢查文件大小并返回一個消息，說明文件已存在。
- 否則，它將寫入文件內容。
- 在這個函數的結尾，使用
run_in_thread_pool方法在多線程環(huán)境中執(zhí)行文件保存操作，并返回每個文件的保存結果
-
upload_docs 函數
- 目的：上傳文檔，并進行選擇性的向量化。
- 輸入：
- 一系列的文件、知識庫名稱、文件處理參數等。
- 主要功能：
- 驗證知識庫名稱的有效性。
- 將上傳的文件保存到磁盤。
- 對保存的文件進行向量化。
- 輸出：一個基礎響應，指示操作是否成功和失敗文件的列表
下面也定義了一個upload_docs函數，區(qū)別在于，此處upload_docs是定義在類 KnowledgeBase 中
函數內部調用self.kb_service.upload_docs(docs)，意味著它在某種程度上是一個代理方法，將上傳任務交給kb_service來完成
相當于為了在KnowledgeBase類的上下文中提供一個簡化的上傳接口。它可能是為了在某種上下文或配置中使用kb_service來上傳文檔
而下面的第二個upload_docs函數可能是實際處理文檔上傳和向量化任務的函數。在實際的應用場景中，KnowledgeBase類的方法可能會調用下面的第二個upload_docs函數來執(zhí)行實際的上傳操作
-
delete_docs 函數
- 目的：從知識庫中刪除指定的文件。
- 輸入：
- 知識庫名稱、待刪除的文件名列表和其他處理參數。
- 主要功能：
- 驗證知識庫名稱的有效性。
- 刪除知識庫中指定的文件。
- 輸出：一個基礎響應，指示操作是否成功和失敗文件的列表。
-
update_docs函數- 目的：更新知識庫中的文檔。
- 輸入：
- 知識庫名稱、待更新的文件名列表和其他處理參數。
- 主要功能：
- 從文件生成文檔并進行向量化
- 更新知識庫中的指定文件，即將文件添加到知識庫文件列表中
- 輸出：一個基礎響應，指示操作是否成功和失敗文件的列表
其中的KBServiceFactory和KnowledgeFile將分別在下文的：“4.1.2 KBServiceFactory的實現：knowledge_base/kb_service/base.py”和“4.2.2 KnowledgeFile的實現”中介紹
def update_docs( knowledge_base_name: str = Body(..., description="知識庫名稱", examples=["samples"]), file_names: List[str] = Body(..., description="文件名稱，支持多文件", examples=["file_name"]), chunk_size: int = Body(CHUNK_SIZE, description="知識庫中單段文本最大長度"), chunk_overlap: int = Body(OVERLAP_SIZE, description="知識庫中相鄰文本重合長度"), zh_title_enhance: bool = Body(ZH_TITLE_ENHANCE, description="是否開啟中文標題加強"), override_custom_docs: bool = Body(False, description="是否覆蓋之前自定義的docs"), docs: Json = Body({}, description="自定義的docs", examples=[{"test.txt": [Document(page_content="custom doc")]}]), not_refresh_vs_cache: bool = Body(False, description="暫不保存向量庫（用于FAISS）") ) -> BaseResponse: ''' 更新知識庫文檔 ''' # 驗證知識庫名稱 if not validate_kb_name(knowledge_base_name): return BaseResponse(code=403, msg="Don't attack me") # 獲取知識庫服務 kb = KBServiceFactory.get_service_by_name(knowledge_base_name) if kb is None: return BaseResponse(code=404, msg=f"未找到知識庫 {knowledge_base_name}") failed_files = {} kb_files = [] # 生成需要加載docs的文件列表 for file_name in file_names: # 獲取文件詳情 file_detail = get_file_detail(kb_name=knowledge_base_name, filename=file_name) # 如果該文件之前使用了自定義docs，則根據參數決定略過或覆蓋 if file_detail.get("custom_docs") and not override_custom_docs: continue if file_name not in docs: try: # 將文件名和知識庫名組合為一個KnowledgeFile對象，并添加到kb_files列表中 kb_files.append(KnowledgeFile(filename=file_name, knowledge_base_name=knowledge_base_name)) except Exception as e: # 記錄失敗的文件和錯誤信息 msg = f"加載文檔 {file_name} 時出錯：{e}" logger.error(f'{e.__class__.__name__}: {msg}', exc_info=e if log_verbose else None) failed_files[file_name] = msg # 從文件生成docs，并進行向量化 for status, result in files2docs_in_thread(kb_files, chunk_size=chunk_size, chunk_overlap=chunk_overlap, zh_title_enhance=zh_title_enhance): if status: # 成功處理文件后，更新知識庫中的文檔 kb_name, file_name, new_docs = result kb_file = KnowledgeFile(filename=file_name, knowledge_base_name=knowledge_base_name) kb_file.splited_docs = new_docs kb.update_doc(kb_file, not_refresh_vs_cache=True) else: # 記錄失敗的文件和錯誤信息 kb_name, file_name, error = result failed_files[file_name] = error # 將自定義的docs進行向量化 for file_name, v in docs.items(): try: # 對于v中的每個條目，檢查它是否已經是Document類型 # 如果不是，那么將其轉換為Document對象 v = [x if isinstance(x, Document) else Document(**x) for x in v] # 根據文件名和知識庫名稱創(chuàng)建KnowledgeFile對象 kb_file = KnowledgeFile(filename=file_name, knowledge_base_name=knowledge_base_name) # 在知識庫中更新該文件的文檔 kb.update_doc(kb_file, docs=v, not_refresh_vs_cache=True) except Exception as e: # 當遇到異常時，構建一個錯誤消息并使用logger進行記錄 msg = f"為 {file_name} 添加自定義docs時出錯：{e}" logger.error(f'{e.__class__.__name__}: {msg}', exc_info=e if log_verbose else None) failed_files[file_name] = msg # 如果需要刷新向量庫，則進行保存 if not not_refresh_vs_cache: kb.save_vector_store() # 返回響應，包括失敗文件列表 return BaseResponse(code=200, msg=f"更新文檔完成", data={"failed_files": failed_files})
-
download_doc 函數
- 目的：從知識庫中下載指定的文檔。
- 輸入：
- 知識庫名稱、待下載的文件名和其他參數。
- 主要功能：
- 提供文件預覽或下載功能。
- 輸出：一個文件響應，允許用戶下載或預覽文件。
-
recreate_vector_store 函數
- 目的：從內容重建向量存儲。
- 輸入：
- 知識庫名稱、向量存儲類型和其他處理參數。
- 主要功能：
- 清除現有的向量存儲。
- 從文件夾中列出文件。
- 為每個文件生成文檔并加入到知識庫中。
- 輸出：一個流響應，用于實時更新操作進度
總體來說，這段代碼主要為知識庫文檔提供了CRUD操作（創(chuàng)建、讀取、更新、刪除）及相關的向量化處理
4.1.2?KBServiceFactory的實現：knowledge_base/kb_service/base.py
-
導入模塊：
- 基本的Python庫如os, operator。
- 用于數據操作和向量化的庫如numpy, sklearn。
- 項目內部的模塊如langchain.embeddings.base, langchain.docstore.document等。
-
SupportedVSType 類：
- 這是一個簡單的類，用于定義支持的向量存儲類型。例如：FAISS, MILVUS等。
-
KBService 類：
- 這是一個抽象基類，定義了知識庫服務的基本功能和行為。
- 初始化函數：給定一個知識庫名和嵌入模型名稱，進行初始化。
- 提供了一系列方法，如
create_kb (創(chuàng)建知識庫)其中的add_kb_to_db將在下文的“4.3.1 knowledge_base_repository.py：實現add_kb_to_db”中分析# 創(chuàng)建知識庫方法 def create_kb(self): # 檢查doc_path路徑是否存在 if not os.path.exists(self.doc_path): # 如果不存在，創(chuàng)建該目錄 os.makedirs(self.doc_path) # 調用子類中定義的do_create_kb方法來執(zhí)行具體的知識庫創(chuàng)建過程 self.do_create_kb() # 將新的知識庫添加到數據庫中，并返回操作狀態(tài) status = add_kb_to_db(self.kb_name, self.vs_type(), self.embed_model) # 返回操作狀態(tài) return status
add_doc (向知識庫添加文件)其中add_file_to_db將在下文的“4.3.2 knowledge_file_repository.py：實現add_file_to_db/add_docs_to_db”中分析# 向知識庫添加文件方法 def add_doc(self, kb_file: KnowledgeFile, docs: List[Document] = [], **kwargs): # 判斷docs列表是否有內容 if docs: # 設置一個標志，表示這是自定義文檔列表 custom_docs = True # 遍歷傳入的文檔列表 for doc in docs: # 為每個文檔的metadata設置默認的"source"屬性，值為文件的路徑 doc.metadata.setdefault("source", kb_file.filepath) else: # 如果沒有提供docs，從kb_file中讀取文檔內容 docs = kb_file.file2text() # 設置一個標志，表示這不是自定義文檔列表 custom_docs = False # 如果docs列表有內容 if docs: # 刪除與kb_file相關聯的現有文檔 self.delete_doc(kb_file) # 調用子類中定義的do_add_doc方法來執(zhí)行具體的文檔添加過程，并返回文檔信息 doc_infos = self.do_add_doc(docs, **kwargs) # 將新的文檔信息添加到數據庫中，并返回操作狀態(tài) status = add_file_to_db(kb_file, custom_docs=custom_docs, docs_count=len(docs), doc_infos=doc_infos) else: # 如果docs列表為空，則設置操作狀態(tài)為False status = False # 返回操作狀態(tài) return status - 還有一些抽象方法，如do_create_kb、do_search等，子類必須實現這些方法
-
KBServiceFactory 類：
- 這是一個工廠類，根據提供的向量存儲類型返回相應的知識庫服務實例
- 使用靜態(tài)方法獲取不同的知識庫服務實例
# 知識庫服務工廠類 class KBServiceFactory: # 根據向量存儲類型返回相應的知識庫服務實例 @staticmethod def get_instance(vs_type: str, knowledge_base_name: str) -> KBService: if vs_type == SupportedVSType.FAISS: from server.knowledge_base.kb_faiss import KBServiceFaiss return KBServiceFaiss(knowledge_base_name) elif vs_type == SupportedVSType.MILVUS: from server.knowledge_base.kb_milvus import KBServiceMilvus return KBServiceMilvus(knowledge_base_name) else: raise ValueError(f"Unsupported VS type: {vs_type}")
-
get_kb_details 函數：
- 獲取目錄和數據庫中的所有知識庫的詳細信息，并將這些信息合并
-
get_kb_file_details 函數：
- 為指定的知識庫獲取目錄和數據庫中的所有文件的詳細信息，并將這些信息合并。
-
EmbeddingsFunAdapter 類：
- 這是一個適配器類，用于在Embeddings類上添加額外的功能。
- embed_documents 和 embed_query 方法對輸入的文本進行嵌入，并將結果進行標準化。
- aembed_documents 和 aembed_query 是它們的異步版本。
-
score_threshold_process 函數：
- 這是一個簡單的函數，用于在得分低于某個閾值的情況下篩選文檔，并返回前k個文檔
4.2 server/db/models文件夾的更新
4.2.1?KnowledgeBaseModel的實現
server/db/models/knowledge_base_model.py中實現了
from sqlalchemy import Column, Integer, String, DateTime, func
from server.db.base import Base
class KnowledgeBaseModel(Base):
"""
知識庫模型
"""
__tablename__ = 'knowledge_base'
id = Column(Integer, primary_key=True, autoincrement=True, comment='知識庫ID')
kb_name = Column(String(50), comment='知識庫名稱')
vs_type = Column(String(50), comment='向量庫類型')
embed_model = Column(String(50), comment='嵌入模型名稱')
file_count = Column(Integer, default=0, comment='文件數量')
create_time = Column(DateTime, default=func.now(), comment='創(chuàng)建時間')
def __repr__(self):
return f"<KnowledgeBase(id='{self.id}', kb_name='{self.kb_name}', vs_type='{self.vs_type}', embed_model='{self.embed_model}', file_count='{self.file_count}', create_time='{self.create_time}')>"4.2.2?KnowledgeFile的實現
經過仔細查找發(fā)現，在server/db/models /knowledge_file_model.py項目文件中實現了KnowledgeFile
# 導入sqlalchemy所需的模塊和函數
from sqlalchemy import Column, Integer, String, DateTime, Float, Boolean, JSON, func
# 從server.db.base導入Base類，這通常用于ORM的基礎模型
from server.db.base import Base
# 定義KnowledgeFileModel類，用于映射“知識文件”數據模型
class KnowledgeFileModel(Base):
"""
知識文件模型
"""
__tablename__ = 'knowledge_file'
id = Column(Integer, primary_key=True, autoincrement=True, comment='知識文件ID')
file_name = Column(String(255), comment='文件名')
file_ext = Column(String(10), comment='文件擴展名')
kb_name = Column(String(50), comment='所屬知識庫名稱')
document_loader_name = Column(String(50), comment='文檔加載器名稱')
text_splitter_name = Column(String(50), comment='文本分割器名稱')
file_version = Column(Integer, default=1, comment='文件版本')
file_mtime = Column(Float, default=0.0, comment="文件修改時間")
file_size = Column(Integer, default=0, comment="文件大小")
custom_docs = Column(Boolean, default=False, comment="是否自定義docs")
docs_count = Column(Integer, default=0, comment="切分文檔數量")
create_time = Column(DateTime, default=func.now(), comment='創(chuàng)建時間')
# 定義對象的字符串表示形式
def __repr__(self):
return f"<KnowledgeFile(id='{self.id}', file_name='{self.file_name}',
file_ext='{self.file_ext}', kb_name='{self.kb_name}',
document_loader_name='{self.document_loader_name}',
text_splitter_name='{self.text_splitter_name}',
file_version='{self.file_version}', create_time='{self.create_time}')>"
# 定義FileDocModel類，用于映射“文件-向量庫文檔”數據模型
class FileDocModel(Base):
"""
文件-向量庫文檔模型
"""
# 定義表名為'file_doc'
__tablename__ = 'file_doc'
# 定義id字段為主鍵，并設置自動遞增，并且附加注釋
id = Column(Integer, primary_key=True, autoincrement=True, comment='ID')
# 定義知識庫名稱字段，并附加注釋
kb_name = Column(String(50), comment='知識庫名稱')
# 定義文件名稱字段，并附加注釋
file_name = Column(String(255), comment='文件名稱')
# 定義向量庫文檔ID字段，并附加注釋
doc_id = Column(String(50), comment="向量庫文檔ID")
# 定義元數據字段，默認為一個空字典
meta_data = Column(JSON, default={})
# 定義對象的字符串表示形式
def __repr__(self):
return f"<FileDoc(id='{self.id}', kb_name='{self.kb_name}', file_name='{self.file_name}', doc_id='{self.doc_id}', metadata='{self.metadata}')>"
4.3?server/db/repository
4.3.1?knowledge_base_repository.py：實現add_kb_to_db
def add_kb_to_db(session, kb_name, vs_type, embed_model):
# 查詢指定名稱的知識庫是否存在于數據庫中
kb = session.query(KnowledgeBaseModel).filter_by(kb_name=kb_name).first()
# 如果指定的知識庫不存在，則創(chuàng)建一個新的知識庫實例
if not kb:
kb = KnowledgeBaseModel(kb_name=kb_name, vs_type=vs_type, embed_model=embed_model)
# 將新的知識庫實例添加到session，這樣可以在之后提交到數據庫
session.add(kb)
else: # 如果知識庫已經存在，則更新它的vs_type和embed_model
kb.vs_type = vs_type
kb.embed_model = embed_model
# 返回True，表示操作成功完成
return True至于其中的KnowledgeBaseModel方法，已經在上文的“4.2.1 KnowledgeBaseModel的實現”中分析了
4.3.2 knowledge_file_repository.py：實現add_file_to_db/add_docs_to_db
- add_file_to_db，將添加文件到數據庫，最后會調用add_docs_to_db
- add_docs_to_db，將文檔添加到數據庫
# 定義向數據庫添加文件的函數
def add_file_to_db(session, # 數據庫會話對象
kb_file: KnowledgeFile, # 知識文件對象
docs_count: int = 0, # 文檔數量，默認為0
custom_docs: bool = False, # 是否為自定義文檔，默認為False
doc_infos: List[str] = [], # 文檔信息列表，默認為空。形式為：[{"id": str, "metadata": dict}, ...]
):
# 從數據庫中查詢與知識庫名相匹配的知識庫記錄
kb = session.query(KnowledgeBaseModel).filter_by(kb_name=kb_file.kb_name).first()
# 如果該知識庫存在
if kb:
# 查詢與文件名和知識庫名相匹配的文件記錄
existing_file: KnowledgeFileModel = (session.query(KnowledgeFileModel)
.filter_by(file_name=kb_file.filename,
kb_name=kb_file.kb_name)
.first())
# 獲取文件的修改時間
mtime = kb_file.get_mtime()
# 獲取文件的大小
size = kb_file.get_size()
# 如果該文件已存在
if existing_file:
# 更新文件的修改時間
existing_file.file_mtime = mtime
# 更新文件的大小
existing_file.file_size = size
# 更新文檔數量
existing_file.docs_count = docs_count
# 更新自定義文檔標志
existing_file.custom_docs = custom_docs
# 文件版本號自增
existing_file.file_version += 1
# 如果文件不存在
else:
# 創(chuàng)建一個新的文件記錄對象
new_file = KnowledgeFileModel(
file_name=kb_file.filename,
file_ext=kb_file.ext,
kb_name=kb_file.kb_name,
document_loader_name=kb_file.document_loader_name,
text_splitter_name=kb_file.text_splitter_name or "SpacyTextSplitter",
file_mtime=mtime,
file_size=size,
docs_count = docs_count,
custom_docs=custom_docs,
)
# 知識庫的文件計數增加
kb.file_count += 1
# 將新文件添加到數據庫會話中
session.add(new_file)
# 添加文檔到數據庫
add_docs_to_db(kb_name=kb_file.kb_name, file_name=kb_file.filename, doc_infos=doc_infos)
# 返回True表示操作成功
return True通過查看上面的倒數第二行代碼可知，add_file_to_db最后調用add_docs_to_db以實現添加文檔到數據庫
def add_docs_to_db(session,
kb_name: str,
file_name: str,
doc_infos: List[Dict]):
'''
將某知識庫某文件對應的所有Document信息添加到數據庫
doc_infos形式：[{"id": str, "metadata": dict}, ...]
'''
for d in doc_infos:
obj = FileDocModel(
kb_name=kb_name,
file_name=file_name,
doc_id=d["id"],
meta_data=d["metadata"],
)
session.add(obj)
return True?更多暫先課上見：七月LLM與langchain/知識圖譜/數據庫的實戰(zhàn) [解決問題、實用為王]，再之后繼續(xù)更新本文
第五部分?langchain-chatchat的二次開發(fā)：商用時的典型問題及其改進方案
上述這個langchain-chatchat開源項目雖好，但真正落地商用時，會遇到各種工程問題，包括且不限于
- 如何解決檢索出錯：embedding算法是關鍵
- 如何解決檢索對了但不根據知識庫回答而是根據模型自有的預訓練知識回答
- 如何針對結構化文檔采取更好的chunk分割：基于規(guī)則
- 如何解決非結構化文檔分割不夠準確的問題：比如最好按照語義切分
- 如何確保召回結果的全面性與準確性：多路召回與最后的去重/精排
- 如何解決基于文檔中表格的問答
- ..
以上內容，請詳見《知識庫問答Langchain-Chatchat的二次開發(fā)：商用時的典型問題及其改進方案》文章來源：http://www.zghlxwxcb.cn/news/detail-523086.html
參考文獻與推薦閱讀
- langchain官網：LangChain、https://python.langchain.com/，API列表：https://api.python.langchain.com/en/latest/api_reference.html
langchain中文網(翻譯暫不佳) - LangChain全景圖
- 一文搞懂langchain(忽略本標題，因為單看此文還不夠)
- How to Build a Smart Chatbot in 10 mins with LangChain
- 關于FAISS的幾篇教程：Faiss入門及應用經驗記錄
- QLoRA：4-bit級別的量化+LoRA方法，用3090在DB-GPT上打造基于33B LLM的個人知識庫
- 基于LangChain+LLM構建增強QA、用LangChain構建大語言模型應用、LangChain 是什么
- 深入剖析開源大模型+Langchain框架智能問答系統(tǒng)性能下降原因
- 七月LLM與langchain/知識圖譜/數據庫的實戰(zhàn) [解決問題、實用為王]
- 深入剖析開源大模型+Langchain框架智能問答系統(tǒng)性能下降原因
后記
本文經歷了三個階段文章來源地址http://www.zghlxwxcb.cn/news/detail-523086.html
- 對langchain的梳理
langchain的組件很多，想理解透徹的話，需要一步步來
包括我自己剛開始看這個庫的時候 真心是暈，無從下手，后來10天過后，可以直接一個文件一個文件的點開 直接看..
總之，凡事都是一個過程 - 對langchain-ChatGLM項目源碼的解讀
說實話，一開始也是挺暈的，因為各種項目文件又很多，好在后來歷時一周總算梳理清楚了 - 開寫新的：第四部分 23年9月升級版Langchain-Chatchat的源碼解析
創(chuàng)作、修改、優(yōu)化記錄
- 7.5-7.9日，每天寫一一部分
- 7.10，完善第一部分關于什么是langchain的介紹
- 7.11，根據langchain-ChatGLM項目的最新更新，整理已寫內容
- 7.12 寫完前3.8節(jié)，且根據項目流程調整各個文件夾的解讀順序
相當于歷時近一周，總算把 “l(fā)angchain-ChatGLM的整體代碼架構” 梳理清楚了 - 7.15，補充langchain架構相關的內容，且為方便理解，把整個langchain庫劃分為三個大層：基礎層、能力層、應用層
- 7.17，開始寫第四部分，重點是4.2節(jié)：用知識圖譜增強 LLM的預訓練、推理、可解釋性
- 7.26，續(xù)寫第四部分，開始更新第五部分：LLM與數據庫的結合
- 9.15，把原來的第四部分、第五部分抽取出來獨立成一篇新的文章：知識圖譜通俗導論：從什么是KG到LLM與KG/DB的結合實戰(zhàn)
- 9.16，開寫新的：第四部分 23年9月升級版Langchain-Chatchat的源碼解析
- 10.26，補充針對“faiss庫的search函數”的細致說明及代碼實現，這點對理解查找K個相似的向量等相關流程很重要
- 11.8，因得知好幾個大廠的AI部門在密切關注本文的更新進度，故更新了二次開發(fā)時遇到的幾個問題作為示例
- 12.27，補充為何要進行切割的原因，以及補充一個 embedding 的 benchmark
到了這里，關于給LLM裝上知識：從LangChain+LLM的本地知識庫問答到LLM與知識圖譜的結合的文章就介紹完了。如果您還想了解更多內容，請在右上角搜索TOY模板網以前的文章或繼續(xù)瀏覽下面的相關文章，希望大家以后多多支持TOY模板網！
