LLM大模型教程：langchain 教程

news2025/3/17 14:08:14

软件安装

pip install pymupdf
pip install langchain
pip install langchain-cli
conda install -c pytorch -c nvidia faiss-gpu=1.7.4 mkl=2021 blas=1.0=mkl

由于langchain不支持qwen模型，我们需要自定义模型

from typing import Any

from langchain_core.language_models import LLM
from transformers import AutoTokenizer, AutoModelForCausalLM

class QwenLLM(LLM):
    model: Any = None
    tokenizer: Any = None
    device: str = None

    @classmethod
    def from_model_id(
            cls,
            model_id: str):
        model = AutoModelForCausalLM.from_pretrained(
            model_id,
            torch_dtype="auto",
            trust_remote_code=True,
            device_map="auto"
        )
        device = "cuda"
        tokenizer = AutoTokenizer.from_pretrained(model_id, trust_remote_code=True)
        return cls(
            model=model,
            device=device,
            tokenizer=tokenizer
        )

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

    def _call(self, prompt: str, **kwargs: Any) -> str:
        messages = [
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": prompt}
        ]

        text = self.tokenizer.apply_chat_template(
            messages,
            tokenize=False,
            add_generation_prompt=True
        )
        model_inputs = self.tokenizer([text], return_tensors="pt").to(self.device)

        generated_ids = self.model.generate(
            model_inputs.input_ids,
            max_new_tokens=512
        )
        generated_ids = [
            output_ids[len(input_ids):] for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)
        ]

        return self.tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]

langchain 查询代码

from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import PyMuPDFLoader
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.prompts import MessagesPlaceholder
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.chains import create_retrieval_chain
from langchain_core.messages import HumanMessage, AIMessage
from qwen_llm import QwenLLM

def main():
    llm = QwenLLM.from_model_id(
        model_id="/home/chuan/models/qwen/Qwen1___5-7B-Chat-GPTQ-Int4")

    loader = PyMuPDFLoader("./汽车知识.pdf")
    docs = loader.load()

    embeddings = HuggingFaceEmbeddings(model_name="/home/chuan/models/BAAI/bge-m3")

    text_splitter = RecursiveCharacterTextSplitter(chunk_size=256)
    documents = text_splitter.split_documents(docs)
    vector = FAISS.from_documents(documents, embeddings)

    prompt = ChatPromptTemplate.from_template("""
    请根据提供的知识回答问题:
    <知识>
    {context}
    </知识>
    问题: {input}""")

    document_chain = create_stuff_documents_chain(llm=llm, prompt=prompt)

    retriever = vector.as_retriever()
    retrieval_chain = create_retrieval_chain(retriever, document_chain)

    response1 = retrieval_chain.invoke({"input": "什么是自动驾驶?"})
    print(response1["answer"])

    prompt = ChatPromptTemplate.from_messages([
        ("system", "请根据提供的知识回答问题:\n\n{context}"),
        MessagesPlaceholder(variable_name="chat_history"),
        ("user", "{input}"),
    ])
    document_chain = create_stuff_documents_chain(llm=llm, prompt=prompt)
    retrieval_chain = create_retrieval_chain(retriever, document_chain)

    chat_history = [HumanMessage(content="什么是自动驾驶?"), AIMessage(content=response1["answer"])]
    response2 = retrieval_chain.invoke({
        "chat_history": chat_history,
        "input": "怎么进行自动驾驶呢？"
    })
    print(response2["answer"])

if __name__ == '__main__':
    main()

整体流程：

初始化langchain 模型
加载知识库文档
分割文档
加载向量数据库
存储向量数据库
准备prompt
创建 chain
调用chain的invoke 方法

langchain+chatglm.png

整个流程如图所示，图片来源于：github.com/chatchat-sp…

LCEL（Lang Chain Expression Language）的原理

在上面的例子中，我们最后生成的retrieval_chain。

document_chain的初始化

RunnablePassthrough.assign(**{DOCUMENTS_KEY: format_docs}).with_config(
        run_name="format_inputs"
    )
| prompt
| llm
| _output_parser

在上面的代码中，我们总共组合了几种链的结构，即LCEL，为什么langchain可以用链式结构，是因为

在python中，运算符 | 实际上调用的是__or__方法

先看一下我们自定的QwenLLM的类关系图

在父类方法中，可以看到 __or__方法的源码

class Runnable(Generic[Input, Output], ABC):
    def __or__(
        self,
        other: Union[
            Runnable[Any, Other],
            Callable[[Any], Other],
            Callable[[Iterator[Any]], Iterator[Other]],
            Mapping[str, Union[Runnable[Any, Other], Callable[[Any], Other], Any]],
        ],
    ) -> RunnableSerializable[Input, Other]:
        """Compose this runnable with another object to create a RunnableSequence."""
        return RunnableSequence(self, coerce_to_runnable(other))

class RunnableSequence(RunnableSerializable[Input, Output]):  
    def __or__(
        self,
        other: Union[
            Runnable[Any, Other],
            Callable[[Any], Other],
            Callable[[Iterator[Any]], Iterator[Other]],
            Mapping[str, Union[Runnable[Any, Other], Callable[[Any], Other], Any]],
        ],
    ) -> RunnableSerializable[Input, Other]:
        if isinstance(other, RunnableSequence):
            return RunnableSequence(
                self.first,
                *self.middle,
                self.last,
                other.first,
                *other.middle,
                other.last,
                name=self.name or other.name,
            )
        else:
            return RunnableSequence(
                self.first,
                *self.middle,
                self.last,
                coerce_to_runnable(other),
                name=self.name,
            )
            
    @property
    def steps(self) -> List[Runnable[Any, Any]]:
        """All the runnables that make up the sequence in order."""
        return [self.first] + self.middle + [self.last]

实际上整个链会被添加到steps方法中

实际上我们调用 invoke方法会直接调用所有的steps

def invoke(self, input: Input, config: Optional[RunnableConfig] = None) -> Output:
    try:
        for i, step in enumerate(self.steps):
            input = step.invoke(
                input,
                # mark each step as a child run
                patch_config(
                    config, callbacks=run_manager.get_child(f"seq:step:{i+1}")
                ),
            )

在这里插入图片描述