🌵??目錄??🌵
😋?數據連接封裝?🍔
文檔加載器:Document Loaders
文檔處理器:TextSplitter
?向量數據庫與向量檢索
?總結
🍉?緩存封裝:Memory?🏖?
對話上下文:ConversationBufferMemory
只保留一個窗口的上下文:ConversationBufferWindowMemory?
通過Token數控制上下文長度:ConversationTokenBufferMemory
更多嘗試
總結
🌞?Chain 和 LangChain Expression Language (LCEL)?🔆
Pipeline 式調用 PromptTemplate, LLM 和 OutputParser
用 LCEL 實現 RAG
通過 LCEL 實現 Function Calling
通過 LCEL,還可以實現
????????本文將繼續延續Langchain專欄文章,本文將講解Langchain的數據連接封裝、緩存封裝和LCEL,逐漸深入學習Langchain的高級能力,幫助我們更好更快的接觸大模型。
? ? ? ? 初識Langchain可以看看這篇文章:
直通車:LangChain:大模型框架的深度解析與應用探索-CSDN博客
😋數據連接封裝
????????對外部進行加載,如果你需要可以做一些處理、轉換,可以做embedding然后放在store(向量數據)里,然后通過retrieve訪問向量數據庫形式進行檢索。這是一個外部數據連接進來的一個模塊的劃分和邏輯上 的Pipeline。
文檔加載器:Document Loaders
# pip install pypdf
from langchain_community.document_loaders import PyPDFLoaderloader = PyPDFLoader("2401.12599v1.pdf")
pages = loader.load_and_split()print(pages[0].page_content)
文檔處理器:TextSplitter
? ? ? ? 切割器,按照字符切割。示例代碼,真正要使用實現的粒度要比該示例粒度都要細。
from langchain_community.document_loaders import PyPDFLoaderloader = PyPDFLoader("2401.12599v1.pdf")
pages = loader.load_and_split()print(pages[0].page_content)from langchain.text_splitter import RecursiveCharacterTextSplitter# 文檔切分
text_splitter = RecursiveCharacterTextSplitter(chunk_size=300,chunk_overlap=200, length_function=len,add_start_index=True,
)paragraphs = text_splitter.create_documents([pages[0].page_content])
for para in paragraphs:print(para.page_content)print('-------')
LangChain 的 PDFLoader 和 TextSplitter 實現都比較粗糙,實際生產中不建議使用。
?向量數據庫與向量檢索
? ? ? ? 它封裝了和三方數據庫的鏈接和檢索。因為本身就是一個接口的封裝,比如調用chroma是一套接口協議,調用pytorch是一套接口協議,它都不一樣,通過Langchain封裝了你都可以用同一套接口去訪問向量數據庫了,將來我要換一個向量數據庫,我都不需要大規模的改代碼都能使用同一套接口去實現。
# pip install chromadb
# 加載 .env 到環境變量
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv())from langchain.document_loaders import UnstructuredMarkdownLoader
from langchain_openai import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.vectorstores import Chroma
from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain_community.document_loaders import PyPDFLoader# 加載文檔
loader = PyPDFLoader("2401.12599v1.pdf")
pages = loader.load_and_split()# 文檔切分
text_splitter = RecursiveCharacterTextSplitter(chunk_size=300,chunk_overlap=100,length_function=len,add_start_index=True,
)texts = text_splitter.create_documents([pages[2].page_content,pages[3].page_content])# 灌庫
embeddings = OpenAIEmbeddings()
db = Chroma.from_documents(texts, embeddings)# LangChain內置的 RAG 實現
qa_chain = RetrievalQA.from_chain_type(llm=ChatOpenAI(temperature=0),retriever=db.as_retriever()
)query = "llm基于哪些數據來訓練?"
response = qa_chain.invoke(query)
print(response["result"])# LLM(Large Language Models)基于大量的文本數據進行訓練。這些數據可以包括互聯網上的網頁、書籍、新聞文章、論文等各種文本資源。通過對這些數據進行深度學習訓練,LLM可以學習到豐富的語言知識和模式,從而具備處理和生成文本的能力。
更多的三方檢索組件鏈接,參考:Vector stores | 🦜?🔗 LangChain
?總結
- 文檔處理部分 LangChain 實現較為粗糙,實際生產中不建議使用
- 與向量數據庫的鏈接部分本質是接口封裝,向量數據庫需要自己選型
🍉緩存封裝:Memory
對話上下文:ConversationBufferMemory
? ? ? ? 它的一個特點是通過一個Buffer存儲上下文,并且可以存儲多行會話,當需要展示歷史會話是可以按寫入順序全部打印出來。
from langchain.memory import ConversationBufferMemory, ConversationBufferWindowMemoryhistory = ConversationBufferMemory()
history.save_context({"input": "hello"}, {"output": "hello"})
print(history.load_memory_variables({}))
history.save_context({"input": "nice to meet you"}, {"output": "nice to meet you too"})
print(history.load_memory_variables({}))
history.save_context({"input": "good bye"}, {"output": "good bye"})
print(history.load_memory_variables({})) # {'history': 'Human: hello\nAI: hello'}
# {'history': 'Human: hello\nAI: hello\nHuman: nice to meet you\nAI: nice to meet you too'}
# {'history': 'Human: hello\nAI: hello\nHuman: nice to meet you\nAI: nice to meet you too\nHuman: good bye\nAI: bye'}
只保留一個窗口的上下文:ConversationBufferWindowMemory?
? ? ? ? 與ConversationBufferMemory不同的時候它能根據預設的會話大小進行截斷,不會造成因為會話太多導致報錯。這種只能保證輪數,但不能保證不會報錯。
from langchain.memory import ConversationBufferWindowMemorywindow = ConversationBufferWindowMemory(k=2)# 保留多少輪問答
window.save_context({"input": "hello"}, {"output": "hello"})
print(window.load_memory_variables({}))
window.save_context({"input": "nice to meet you"}, {"output": "nice to meet you too"})
print(window.load_memory_variables({}))
window.save_context({"input": "how old are you?"}, {"output": "#!@!#!!¥"})
print(window.load_memory_variables({}))
window.save_context({"input": "good bye"}, {"output": "bye"})
print(window.load_memory_variables({}))# {'history': 'Human: hello\nAI: hello'}
# {'history': 'Human: hello\nAI: hello\nHuman: nice to meet you\nAI: nice to meet you too'}
# {'history': 'Human: nice to meet you\nAI: nice to meet you too\nHuman: how old are you?# \nAI: #!@!#!!¥'}
# {'history': 'Human: how old are you?\nAI: #!@!#!!¥\nHuman: good bye\nAI: bye'}
通過Token數控制上下文長度:ConversationTokenBufferMemory
? ? ? ? 能夠限制Memory最多能存多少個Token,超過Token就遺棄,它可以保證調用會話不會報錯。
# 加載 .env 到環境變量
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv())from langchain.memory import ConversationTokenBufferMemory
from langchain_openai import ChatOpenAImemory = ConversationTokenBufferMemory(llm=ChatOpenAI(),# max_token_limit=40
)
memory.save_context({"input": "你好啊"}, {"output": "你好,我是你的AI助手。"})
memory.save_context({"input": "你會干什么"}, {"output": "我什么都會"})print(memory.load_memory_variables({}))
# {'history': 'Human: 你會干什么\nAI: 我什么都會'}
更多嘗試
ConversationSummaryMemory: 對上下文做摘要
傳送門:Conversation Summary | 🦜?🔗 LangChain
ConversationSummaryBufferMemory: 保存 Token 數限制內的上下文,對更早的做摘要
傳送門:Conversation Summary Buffer | 🦜?🔗 LangChain
VectorStoreRetrieverMemory: 將 Memory 存儲在向量數據庫中,根據用戶輸入檢索回最相關的部分
傳送門:Backed by a Vector Store | 🦜?🔗 LangChain
總結
- LangChain 的 Memory 管理機制屬于可用的部分,尤其是簡單情況如按輪數或按 Token 數管理;
- 對于復雜情況,它不一定是最優的實現,例如檢索向量庫方式,建議根據實際情況和效果評估;
- 但是它對內存的各種維護方法的思路在實際生產中可以借鑒。
🌞Chain 和 LangChain Expression Language (LCEL)
????????LangChain Expression Language(LCEL)是一種聲明式語言,可輕松組合不同的調用順序構成 Chain。LCEL 自創立之初就被設計為能夠支持將原型投入生產環境,無需代碼更改,從最簡單的“提示+LLM”鏈到最復雜的鏈(已有用戶成功在生產環境中運行包含數百個步驟的 LCEL Chain)。
LCEL 的一些亮點包括:
-
流支持:使用 LCEL 構建 Chain 時,你可以獲得最佳的首個令牌時間(即從輸出開始到首批輸出生成的時間)。對于某些 Chain,這意味著可以直接從 LLM 流式傳輸令牌到流輸出解析器,從而以與 LLM 提供商輸出原始令牌相同的速率獲得解析后的、增量的輸出。
-
異步支持:任何使用 LCEL 構建的鏈條都可以通過同步 API(例如,在 Jupyter 筆記本中進行原型設計時)和異步 API(例如,在 LangServe 服務器中)調用。這使得相同的代碼可用于原型設計和生產環境,具有出色的性能,并能夠在同一服務器中處理多個并發請求。
-
優化的并行執行:當你的 LCEL 鏈條有可以并行執行的步驟時(例如,從多個檢索器中獲取文檔),我們會自動執行,無論是在同步還是異步接口中,以實現最小的延遲。
-
重試和回退:為 LCEL 鏈的任何部分配置重試和回退。這是使鏈在規模上更可靠的絕佳方式。目前我們正在添加重試/回退的流媒體支持,因此你可以在不增加任何延遲成本的情況下獲得增加的可靠性。
-
訪問中間結果:對于更復雜的鏈條,訪問在最終輸出產生之前的中間步驟的結果通常非常有用。這可以用于讓最終用戶知道正在發生一些事情,甚至僅用于調試鏈條。你可以流式傳輸中間結果,并且在每個 LangServe 服務器上都可用。
-
輸入和輸出模式:輸入和輸出模式為每個 LCEL 鏈提供了從鏈的結構推斷出的 Pydantic 和 JSONSchema 模式。這可以用于輸入和輸出的驗證,是 LangServe 的一個組成部分。
-
無縫 LangSmith 跟蹤集成:隨著鏈條變得越來越復雜,理解每一步發生了什么變得越來越重要。通過 LCEL,所有步驟都自動記錄到 LangSmith,以實現最大的可觀察性和可調試性。
-
無縫 LangServe 部署集成:任何使用 LCEL 創建的鏈都可以輕松地使用 LangServe 進行部署。?
直通車:LangChain Expression Language (LCEL) | 🦜?🔗 LangChain
?Pipeline 式調用 PromptTemplate, LLM 和 OutputParser
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv())from langchain.output_parsers import PydanticOutputParser
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema import StrOutputParser
from langchain.schema.runnable import RunnablePassthrough
from pydantic import BaseModel, Field, validator
from typing import List, Dict, Optional
from enum import Enum# 輸出結構
class SortEnum(str, Enum):data = 'data'price = 'price'class OrderingEnum(str, Enum):ascend = 'ascend'descend = 'descend'class Semantics(BaseModel):name: Optional[str] = Field(description="流量包名稱",default=None)price_lower: Optional[int] = Field(description="價格下限",default=None)price_upper: Optional[int] = Field(description="價格上限",default=None)data_lower: Optional[int] = Field(description="流量下限",default=None)data_upper: Optional[int] = Field(description="流量上限",default=None)sort_by: Optional[SortEnum] = Field(description="按價格或流量排序",default=None)ordering: Optional[OrderingEnum] = Field(description="升序或降序排列",default=None)# OutputParser
parser = PydanticOutputParser(pydantic_object=Semantics)# Prompt 模板
prompt = ChatPromptTemplate.from_messages([("system","將用戶的輸入解析成JSON表示。輸出格式如下:\n{format_instructions}\n不要輸出未提及的字段。",),("human", "{query}"),]
).partial(format_instructions=parser.get_format_instructions())# 模型
model = ChatOpenAI(temperature=0)# LCEL 表達式
# | 表示從左到右運行,| 前面的操作符是管道符,它表示將前一個操作符的輸出作為下一個操作符的
# 1.拿用戶輸入的填充到模板中
# 2.將 RunnablePassthrough 對象傳遞給 prompt,并返回一個新的 RunnablePassthrough 對象
# 3.將新的 RunnablePassthrough 對象傳遞給 model,并返回一個新的 RunnablePassthrough 對象
# 4.將新的 RunnablePassthrough 對象傳遞給 parser 格式化輸出
runnable = ({"query": RunnablePassthrough()} | prompt | model | parser
)# 運行
print(runnable.invoke("不超過100元的流量大的套餐有哪些"))
注意:?在當前的文檔中 LCEL 產生的對象,被叫做 runnable 或 chain,經常兩種叫法混用。本質就是一個自定義調用流程。
?使用 LCEL 的價值,也就是 LangChain 的核心價值。
官方從不同角度給出了舉例說明:https://python.langchain.com/docs/expression_language/why
通過 LCEL 實現 RAG
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv())from langchain.prompts import ChatPromptTemplate
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_openai import ChatOpenAI
from langchain_community.document_loaders import PyPDFLoader# 模型
model = ChatOpenAI(model="gpt-4-turbo", temperature=0)# 加載文檔
loader = PyPDFLoader("2401.12599v1.pdf")
pages = loader.load_and_split()# 文檔切分
text_splitter = RecursiveCharacterTextSplitter(chunk_size=200,chunk_overlap=100,length_function=len,add_start_index=True,
)texts = text_splitter.create_documents([page.page_content for page in pages[:4]]
)# 灌庫
embeddings = OpenAIEmbeddings(model="text-embedding-ada-002")
db = FAISS.from_documents(texts, embeddings)# 檢索 top-1 結果
retriever = db.as_retriever(search_kwargs={"k": 5})from langchain.schema.output_parser import StrOutputParser
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema.output_parser import StrOutputParser
from langchain.schema.runnable import RunnablePassthrough# Prompt模板
template = """Answer the question based only on the following context:
{context}Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)# Chain
rag_chain = ({"question": RunnablePassthrough(), "context": retriever}| prompt| model| StrOutputParser()
)invoke = rag_chain.invoke("Llama llm基于哪些數據來訓練?")
print(invoke)
# 基于上下文的信息,大型語言模型(Large language models,LLM)主要是根據來自公開可用的互聯網資源訓練的。這些資源包括網頁、書籍、新聞和對話文本。因此,Llama LLM是基于這些類型的數據來訓練的。
通過 LCEL 實現 Function Calling
from dotenv import load_dotenv, find_dotenv
_ = load_dotenv(find_dotenv())from langchain_openai import ChatOpenAI
# 提供了一個注解器
from langchain_core.tools import tool# 模型
model = ChatOpenAI(model="gpt-4-turbo", temperature=0)# 通過這個注解標記一個函數,它將被 LangChain 用來調用和執行。
@tool
def multiply(first_int: int, second_int: int) -> int:"""兩個整數相乘"""return first_int * second_int@tool
def add(first_int: int, second_int: int) -> int:"""兩數之和"""return first_int + second_int@tool
def exponentiate(base: int, exponent: int) -> int:"""乘方"""return base**exponentfrom langchain_core.output_parsers import StrOutputParser
from langchain.output_parsers import JsonOutputToolsParsertools = [multiply, add, exponentiate]# 通過語言模型調用function calling
# 帶有分支的 LCEL
# 它會將你定義的函數,生成openai的schema,然后調用openai的模型
llm_with_tools = model.bind_tools(tools) | {# llm自己去判斷是返回function calling的指示還是返回文本回復"functions": JsonOutputToolsParser(),"text": StrOutputParser()
}# result = llm_with_tools.invoke("1024的16倍是多少")
# result = llm_with_tools.invoke("1+1等于多少")
result = llm_with_tools.invoke("你是誰")
print(result)
# {'functions': [{'args': {'first_int': 1024, 'second_int': 16}, 'type': 'multiply'}], 'text': ''}
# {'functions': [{'args': {'first_int': 1, 'second_int': 1}, 'type': 'add'}], 'text': ''}
# {'functions': [], 'text': '我是一個聊天機器人。我可以回答你的問題,提供幫助和建議。\n我是一個聊天機器人。我可以回答你的問題,提供幫助和建議。'}
注意:如果執行報NotImplementedError錯誤,可以更新langchain版本試試:
python -m pip install --upgrade pip
pip install -qU langchain-openai
更多實現
- 配置運行時變量:Configure runtime chain internals | 🦜?🔗 LangChain
- 故障回退:Fallbacks | 🦜?🔗 LangChain
- 并行調用:Parallel: Format data | 🦜?🔗 LangChain
- 邏輯分支:Route logic based on input | 🦜?🔗 LangChain
- 調用自定義流式函數:Lambda: Run custom functions | 🦜?🔗 LangChain
- 鏈接外部 Memory:Add message history (memory) | 🦜?🔗 LangChain
更多例子:LangChain Expression Language (LCEL) | 🦜?🔗 LangChain?