LLM大模型教程：langchain 教程

本文主要是介绍LLM大模型教程：langchain 教程，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

软件安装

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 Anyfrom langchain_core.language_models import LLM
from transformers import AutoTokenizer, AutoModelForCausalLMclass QwenLLM(LLM):model: Any = Nonetokenizer: Any = Nonedevice: str = None@classmethoddef 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)@propertydef _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 QwenLLMdef 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,)@propertydef 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 runpatch_config(config, callbacks=run_manager.get_child(f"seq:step:{i+1}")),)

在这里插入图片描述