Langchain：数据连接封装、缓存封装和LCEL学习和探索

🌵 目录 🌵

😋 数据连接封装 🍔

文档加载器：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 PyPDFLoader

loader = PyPDFLoader("2401.12599v1.pdf")
pages = loader.load_and_split()

print(pages[0].page_content)

文档处理器：TextSplitter

切割器，按照字符切割。示例代码，真正要使用实现的粒度要比该示例粒度都要细。

from langchain_community.document_loaders import PyPDFLoader

loader = 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, ConversationBufferWindowMemory

history = 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 ConversationBufferWindowMemory

window = 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 ChatOpenAI

memory = 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: 我什么都会'}

总结

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**exponent

from langchain_core.output_parsers import StrOutputParser
from langchain.output_parsers import JsonOutputToolsParser

tools = [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我是一个聊天机器人。我可以回答你的问题，提供帮助和建议。'}