Langchain.rst

Langchain
==========================

.. attention:: 
    To be updated for Qwen3.

This guide helps you build a question-answering application based 
on a local knowledge base using ``Qwen2.5-7B-Instruct`` with ``langchain``.
The goal is to establish a knowledge base Q&A solution.

Basic Usage
-----------

The implementation process of this project includes 
loading files -> reading text -> segmenting text -> vectorizing text -> vectorizing questions 
-> matching the top k most similar text vectors with the question vectors -> 
incorporating the matched text as context along with the question into the prompt -> 
submitting to the Qwen2.5-7B-Instruct to generate an answer.
Below is an example:

.. code:: bash

   pip install langchain==0.0.174
   pip install faiss-gpu

.. code:: python

   from transformers import AutoModelForCausalLM, AutoTokenizer
   from abc import ABC
   from langchain.llms.base import LLM
   from typing import Any, List, Mapping, Optional
   from langchain.callbacks.manager import CallbackManagerForLLMRun
   
   model_name = "Qwen/Qwen2.5-7B-Instruct"

   model = AutoModelForCausalLM.from_pretrained(
       model_name,
       torch_dtype="auto",
       device_map="auto"
   )
   tokenizer = AutoTokenizer.from_pretrained(model_name)

   class Qwen(LLM, ABC):
        max_token: int = 10000
        temperature: float = 0.01
        top_p = 0.9
        history_len: int = 3

        def __init__(self):
            super().__init__()

        @property
        def _llm_type(self) -> str:
            return "Qwen"

        @property
        def _history_len(self) -> int:
            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,
            run_manager: Optional[CallbackManagerForLLMRun] = None,
        ) -> str:
            messages = [
                {"role": "system", "content": "You are Qwen, created by Alibaba Cloud. You are a helpful assistant."},
                {"role": "user", "content": prompt}
            ]
            text = tokenizer.apply_chat_template(
                messages,
                tokenize=False,
                add_generation_prompt=True
            )
            model_inputs = tokenizer([text], return_tensors="pt").to(model.device)
            generated_ids = model.generate(
                **model_inputs,
                max_new_tokens=512
            )
            generated_ids = [
                output_ids[len(input_ids):] for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)
            ]

            response = tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]
            return response
        
        @property
        def _identifying_params(self) -> Mapping[str, Any]:
            """Get the identifying parameters."""
            return {"max_token": self.max_token,
                    "temperature": self.temperature,
                    "top_p": self.top_p,
                    "history_len": self.history_len}

After loading the Qwen2.5-7B-Instruct model, you should specify the txt file 
for retrieval.

.. code:: python

    import os
    import re
    import torch
    import argparse
    from langchain.vectorstores import FAISS
    from langchain.embeddings.huggingface import HuggingFaceEmbeddings
    from typing import List, Tuple
    import numpy as np
    from langchain.document_loaders import TextLoader
    from langchain.text_splitter import CharacterTextSplitter
    from langchain.docstore.document import Document
    from langchain.prompts.prompt import PromptTemplate
    from langchain.chains import RetrievalQA
    
    class ChineseTextSplitter(CharacterTextSplitter):
        def __init__(self, pdf: bool = False, **kwargs):
            super().__init__(**kwargs)
            self.pdf = pdf

        def split_text(self, text: str) -> List[str]:
            if self.pdf:
                text = re.sub(r"\n{3,}", "\n", text)
                text = re.sub('\s', ' ', text)
                text = text.replace("\n\n", "")
            sent_sep_pattern = re.compile(
                '([﹒﹔﹖﹗．。！？]["’”」』]{0,2}|(?=["‘“「『]{1,2}|$))') 
            sent_list = []
            for ele in sent_sep_pattern.split(text):
                if sent_sep_pattern.match(ele) and sent_list:
                    sent_list[-1] += ele
                elif ele:
                    sent_list.append(ele)
            return sent_list


    def load_file(filepath):
        loader = TextLoader(filepath, autodetect_encoding=True)
        textsplitter = ChineseTextSplitter(pdf=False)
        docs = loader.load_and_split(textsplitter)
        write_check_file(filepath, docs)
        return docs


    def write_check_file(filepath, docs):
        folder_path = os.path.join(os.path.dirname(filepath), "tmp_files")
        if not os.path.exists(folder_path):
            os.makedirs(folder_path)
        fp = os.path.join(folder_path, 'load_file.txt')
        with open(fp, 'a+', encoding='utf-8') as fout:
            fout.write("filepath=%s,len=%s" % (filepath, len(docs)))
            fout.write('\n')
            for i in docs:
                fout.write(str(i))
                fout.write('\n')
            fout.close()

            
    def separate_list(ls: List[int]) -> List[List[int]]:
        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 lists


    class FAISSWrapper(FAISS):
        chunk_size = 250
        chunk_conent = True
        score_threshold = 0
        
        def similarity_search_with_score_by_vector(
                self, embedding: List[float], k: int = 4
        ) -> List[Tuple[Document, float]]:
            scores, indices = self.index.search(np.array([embedding], dtype=np.float32), k)
            docs = []
            id_set = set()
            store_len = len(self.index_to_docstore_id)
            for j, i in enumerate(indices[0]):
                if i == -1 or 0 < self.score_threshold < scores[0][j]:
                    # This happens when not enough docs are returned.
                    continue
                _id = self.index_to_docstore_id[i]
                doc = self.docstore.search(_id)
                if not self.chunk_conent:
                    if not isinstance(doc, Document):
                        raise ValueError(f"Could not find document for id {_id}, got {doc}")
                    doc.metadata["score"] = int(scores[0][j])
                    docs.append(doc)
                    continue
                id_set.add(i)
                docs_len = len(doc.page_content)
                for k in range(1, max(i, store_len - i)):
                    break_flag = False
                    for l in [i + k, i - k]:
                        if 0 <= l < len(self.index_to_docstore_id):
                            _id0 = self.index_to_docstore_id[l]
                            doc0 = self.docstore.search(_id0)
                            if docs_len + len(doc0.page_content) > self.chunk_size:
                                break_flag = True
                                break
                            elif doc0.metadata["source"] == doc.metadata["source"]:
                                docs_len += len(doc0.page_content)
                                id_set.add(l)
                    if break_flag:
                        break
            if not self.chunk_conent:
                return docs
            if len(id_set) == 0 and self.score_threshold > 0:
                return []
            id_list = sorted(list(id_set))
            id_lists = separate_list(id_list)
            for id_seq in id_lists:
                for id in id_seq:
                    if id == id_seq[0]:
                        _id = self.index_to_docstore_id[id]
                        doc = self.docstore.search(_id)
                    else:
                        _id0 = self.index_to_docstore_id[id]
                        doc0 = self.docstore.search(_id0)
                        doc.page_content += " " + doc0.page_content
                if not isinstance(doc, Document):
                    raise ValueError(f"Could not find document for id {_id}, got {doc}")
                doc_score = min([scores[0][id] for id in [indices[0].tolist().index(i) for i in id_seq if i in indices[0]]])
                doc.metadata["score"] = int(doc_score)
                docs.append((doc, doc_score))
            return docs


    if __name__ == '__main__':
        # load docs (pdf file or txt file)
        filepath = 'your file path'
        # Embedding model name
        EMBEDDING_MODEL = 'text2vec'
        PROMPT_TEMPLATE = """Known information:
        {context_str} 
        Based on the above known information, respond to the user's question concisely and professionally. If an answer cannot be derived from it, say 'The question cannot be answered with the given information' or 'Not enough relevant information has been provided,' and do not include fabricated details in the answer. Please respond in English. The question is {question}"""
        # Embedding running device
        EMBEDDING_DEVICE = "cuda"
        # return top-k text chunk from vector store
        VECTOR_SEARCH_TOP_K = 3
        CHAIN_TYPE = 'stuff'
        embedding_model_dict = {
            "text2vec": "your text2vec model path",
        }
        llm = Qwen()
        embeddings = HuggingFaceEmbeddings(model_name=embedding_model_dict[EMBEDDING_MODEL],model_kwargs={'device': EMBEDDING_DEVICE})
        
        docs = load_file(filepath)
        
        docsearch = FAISSWrapper.from_documents(docs, embeddings)
        
        prompt = PromptTemplate(
            template=PROMPT_TEMPLATE, input_variables=["context_str", "question"]
        )

        chain_type_kwargs = {"prompt": prompt, "document_variable_name": "context_str"}
        qa = RetrievalQA.from_chain_type(
            llm=llm,
            chain_type=CHAIN_TYPE, 
            retriever=docsearch.as_retriever(search_kwargs={"k": VECTOR_SEARCH_TOP_K}), 
            chain_type_kwargs=chain_type_kwargs)
        
        query = "Give me a short introduction to large language model."  
        print(qa.run(query))

Next Step
---------

Now you can chat with Qwen2.5 use your own document. Continue
to read the documentation and try to figure out more advanced usages of
model retrieval!