【LLM学习笔记4】使用LangChain开发应用程序（上）

前言

在前面两部分，我们分别学习了大语言模型的基础使用准则（Prompt Engineering）与如何基于 ChatGPT 搭建一个完整的问答系统，对基于 LLM 开发应用程序有了一定了解。但是，虽然 LLM 提供了强大的能力，极大便利了应用程序的开发，个人开发者要基于 LLM 快速、便捷地开发一个完整的应用程序依然是一个具有较大工作量的任务。针对 LLM 开发，LangChain 应运而生。LangChain 是一套专为 LLM 开发打造的开源框架，实现了 LLM 多种强大能力的利用，提供了 Chain、Agent、Tool 等多种封装工具，基于 LangChain 可以便捷开发应用程序，极大化发挥 LLM 潜能。目前，使用 LangChain 已经成为 LLM 开发的必备能力之一。

langchain官网文档langchainchina

一、模型、提示和解析器（model、prompt、parsers）

langchain的主要功能：

1. 提供了更结构化的模型调用方式，包括模型(Models)、提示(Prompts)和输出解析器(Output Parsers)三个主要组件
2. 提供了模板化的提示管理系统
3. 提供了输出格式的规范化处理

相比直接调用的优势：

1. 提示模板化：通过ChatPromptTemplate可以更好地管理和重用提示模板
2. 结构化输出：通过OutputParser可以将模型返回的字符串解析成Python对象(如dict)，方便后续处理
3. 统一接口：支持多种模型，便于切换和管理
4. 内置常用场景：提供了许多预设的提示模板，如摘要、问答等

使用方法：

模型调用:

from langchain.chat_models import ChatOpenAI
chat = ChatOpenAI(temperature=0.0)

提示模板:

from langchain.prompts import ChatPromptTemplate
template = ChatPromptTemplate.from_template(template_string)
messages = template.format_messages(variables)

输出解析:

from langchain.output_parsers import ResponseSchema, StructuredOutputParser
# 定义输出schema
schemas = [ResponseSchema(...)]
# 创建解析器
parser = StructuredOutputParser.from_response_schemas(schemas)
# 解析输出
result = parser.parse(response.content)

总的来说，langchain提供了一个更加工程化和结构化的方式来使用大语言模型，特别适合构建生产级别的应用。

二、储存

在与语言模型交互时，你可能已经注意到一个关键问题：它们并不记忆你之前的交流内容，这在我们构建一些应用程序（如聊天机器人）的时候，带来了很大的挑战，使得对话似乎缺乏真正的连续性。因此，在本节中我们将介绍 LangChain 中的储存模块，即如何将先前的对话嵌入到语言模型中的，使其具有连续对话的能力。

当使用 LangChain 中的储存(Memory)模块时，它旨在保存、组织和跟踪整个对话的历史，从而为用户和模型之间的交互提供连续的上下文。

LangChain 提供了多种储存类型。其中，缓冲区储存允许保留最近的聊天消息，摘要储存则提供了对整个对话的摘要。实体储存则允许在多轮对话中保留有关特定实体的信息。这些记忆组件都是模块化的，可与其他组件组合使用，从而增强机器人的对话管理能力。储存模块可以通过简单的 API 调用来访问和更新，允许开发人员更轻松地实现对话历史记录的管理和维护。

langchain中四种主要的储存模块及其特点：

1. 对话缓存储存 (ConversationBufferMemory)

• 功能：完整保存所有历史对话内容
• 特点：
  • 能记住完整对话历史
  • 可以通过save_context直接添加内容
  • 随着对话增加会占用越来越多内存
• 适用场景：需要完整对话历史的短期对话

2. 对话缓存窗口储存 (ConversationBufferWindowMemory)

• 功能：只保留最近k轮对话
• 特点：
  • 通过设置k值控制保留对话轮数
  • 节省内存空间
  • 只能访问最近k轮对话
• 适用场景：只需要最近几轮对话上下文的场合

3. 对话令牌缓存储存 (ConversationTokenBufferMemory)

• 功能：限制保存的token数量
• 特点：
  • 可以设置最大token限制
  • 超出限制时会裁剪早期对话
  • 基于tiktoken计算token
• 适用场景：需要控制token使用量的场合

4. 对话摘要缓存储存 (ConversationSummaryBufferMemory)

• 功能：自动总结历史对话为摘要
• 特点：
  • 使用LLM自动生成对话摘要
  • 可以保留关键信息同时节省空间
  • 摘要会随新对话更新
• 适用场景：长对话场景，需要记住重要信息但不需要完整细节

例子：

# 首先需要安装这些包：
# pip install langchain langchain-community openaifrom langchain_community.chat_models import ChatOpenAI
from langchain.chains import ConversationChain
from langchain.memory import (ConversationBufferMemory,ConversationBufferWindowMemory,ConversationTokenBufferMemory,ConversationSummaryBufferMemory
)# 设置OpenAI API密钥
import os
os.environ["OPENAI_API_KEY"] = "你的API密钥"# 初始化模型
llm = ChatOpenAI(temperature=0.0)# 1. 完整对话储存
memory1 = ConversationBufferMemory()
chain1 = ConversationChain(llm=llm, memory=memory1)
chain1.predict(input="你好，我叫小明")
chain1.predict(input="我的名字是什么？")
print("\n完整储存记忆:", memory1.load_memory_variables({}))# 2. 窗口储存(只保留最后1轮对话)
memory2 = ConversationBufferWindowMemory(k=1)
chain2 = ConversationChain(llm=llm, memory=memory2)
chain2.predict(input="你好，我叫小红")
chain2.predict(input="我的名字是什么？")
print("\n窗口储存记忆:", memory2.load_memory_variables({}))# 3. Token储存(限制token数量)
memory3 = ConversationTokenBufferMemory(llm=llm, max_token_limit=30)
chain3 = ConversationChain(llm=llm, memory=memory3)
chain3.predict(input="你好，我叫小华")
chain3.predict(input="我的名字是什么？")
print("\nToken储存记忆:", memory3.load_memory_variables({}))# 4. 摘要储存
memory4 = ConversationSummaryBufferMemory(llm=llm, max_token_limit=30)
chain4 = ConversationChain(llm=llm, memory=memory4)
chain4.predict(input="你好，我叫小李")
chain4.predict(input="我的名字是什么？")
print("\n摘要储存记忆:", memory4.load_memory_variables({}))

这些储存模块的主要优势是：

1. 让无状态的LLM能够"记住"历史对话
2. 提供不同的记忆管理策略
3. 灵活控制内存使用和token消耗
4. 支持更自然的多轮对话

使用时可以根据具体需求选择合适的储存模块。

三、模型链

链（Chains）通常将大语言模型（LLM）与提示（Prompt）结合在一起，基于此，我们可以对文本或数据进行一系列操作。链（Chains）可以一次性接受多个输入。例如，我们可以创建一个链，该链接受用户输入，使用提示模板对其进行格式化，然后将格式化的响应传递给 LLM 。我们可以通过将多个链组合在一起，或者通过将链与其他组件组合在一起来构建更复杂的链。

1. LLMChain (大语言模型链)

from langchain_community.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.chains import LLMChain# 初始化
llm = ChatOpenAI(temperature=0.0)
prompt = ChatPromptTemplate.from_template("描述制造{product}的公司的最佳名称是什么?")
chain = LLMChain(llm=llm, prompt=prompt)# 运行
result = chain.run("床单")

2. SimpleSequentialChain (简单顺序链)

from langchain.chains import SimpleSequentialChain# 创建两个子链
prompt1 = ChatPromptTemplate.from_template("为{product}公司起名")
chain1 = LLMChain(llm=llm, prompt=prompt1)prompt2 = ChatPromptTemplate.from_template("描述{company_name}公司")
chain2 = LLMChain(llm=llm, prompt=prompt2)# 组合成顺序链
overall_chain = SimpleSequentialChain(chains=[chain1, chain2],verbose=True
)# 运行
result = overall_chain.run("床单")

3. SequentialChain (顺序链)

from langchain.chains import SequentialChain# 创建多个子链
chain1 = LLMChain(llm=llm, prompt=prompt1, output_key="company_name")
chain2 = LLMChain(llm=llm, prompt=prompt2, output_key="description")
chain3 = LLMChain(llm=llm, prompt=prompt3, output_key="slogan")# 组合成顺序链
overall_chain = SequentialChain(chains=[chain1, chain2, chain3],input_variables=["product"],output_variables=["company_name", "description", "slogan"],verbose=True
)# 运行
result = overall_chain({"product": "床单"})

4. MultiPromptChain (路由链)

from langchain.chains.router import MultiPromptChain
from langchain.chains.router.llm_router import LLMRouterChain# 创建目标链
destination_chains = {"数学": LLMChain(llm=llm, prompt=math_prompt),"物理": LLMChain(llm=llm, prompt=physics_prompt)
}# 创建默认链
default_chain = LLMChain(llm=llm, prompt=default_prompt)# 创建路由链
router_chain = LLMRouterChain.from_llm(llm, router_prompt)# 组合成路由链
chain = MultiPromptChain(router_chain=router_chain,destination_chains=destination_chains,default_chain=default_chain,verbose=True
)# 运行
result = chain.run("2+2等于多少?")

主要特点：

1. LLMChain: 最基础的链，将prompt和LLM组合
2. SimpleSequentialChain: 单输入单输出的顺序链
3. SequentialChain: 多输入多输出的顺序链
4. MultiPromptChain: 根据输入内容路由到不同的专门链

使用建议：

• 简单任务用LLMChain
• 需要步骤处理用Sequential Chain
• 需要分类处理用MultiPromptChain
• 注意设置verbose=True可以查看链的执行过程

四、基于文档的问答

使用大语言模型构建一个能够回答关于给定文档和文档集合的问答系统是一种非常实用和有效的应用场景。与仅依赖模型预训练知识不同，这种方法可以进一步整合用户自有数据，实现更加个性化和专业的问答服务。例如,我们可以收集某公司的内部文档、产品说明书等文字资料，导入问答系统中。然后用户针对这些文档提出问题时，系统可以先在文档中检索相关信息，再提供给语言模型生成答案。

这样，语言模型不仅利用了自己的通用知识，还可以充分运用外部输入文档的专业信息来回答用户问题，显著提升答案的质量和适用性。构建这类基于外部文档的问答系统，可以让语言模型更好地服务于具体场景，而不是停留在通用层面。这种灵活应用语言模型的方法值得在实际使用中推广。

基于文档问答的这个过程，我们会涉及 LangChain 中的其他组件，比如：嵌入模型（Embedding Models)和向量储存(Vector Stores)。

1.使用向量存储查询

from langchain.chains import RetrievalQA
from langchain.document_loaders import CSVLoader
from langchain.vectorstores import DocArrayInMemorySearch
from langchain.indexes import VectorstoreIndexCreator# 1. 加载数据
loader = CSVLoader(file_path='clothing_catalog.csv')# 2. 创建向量存储索引
index = VectorstoreIndexCreator(vectorstore_cls=DocArrayInMemorySearch
).from_loaders([loader])# 3. 查询
response = index.query("请列出防晒衬衫")

2. 结合表征模型和向量存储

由于语言模型的上下文长度限制，直接处理长文档具有困难。为实现对长文档的问答，我们可以引入向量嵌入(Embeddings)和向量存储(Vector Store)等技术：

首先，使用文本嵌入(Embeddings)算法对文档进行向量化，使语义相似的文本片段具有接近的向量表示。其次，将向量化的文档切分为小块，存入向量数据库，这个流程正是创建索引(index)的过程。向量数据库对各文档片段进行索引，支持快速检索。这样，当用户提出问题时，可以先将问题转换为向量，在数据库中快速找到语义最相关的文档片段。然后将这些文档片段与问题一起传递给语言模型，生成回答。

通过嵌入向量化和索引技术，我们实现了对长文档的切片检索和问答。这种流程克服了语言模型的上下文限制，可以构建处理大规模文档的问答系统。

from langchain.embeddings import OpenAIEmbeddings
from langchain.chat_models import ChatOpenAI# 1. 加载数据
loader = CSVLoader(file_path='clothing_catalog.csv')
docs = loader.load()# 2. 创建文本向量表征
embeddings = OpenAIEmbeddings()# 3. 创建向量存储
db = DocArrayInMemorySearch.from_documents(docs, embeddings)# 4. 相似度搜索
query = "防晒衬衫"
similar_docs = db.similarity_search(query)# 5. 使用LLM处理搜索结果
llm = ChatOpenAI(temperature=0.0)
combined_docs = "".join([doc.page_content for doc in similar_docs])
response = llm.call_as_llm(f"{combined_docs}\n问题：{query}")

使用检索问答链回答问题

通过LangChain创建一个检索问答链，对检索到的文档进行问题回答。检索问答链的输入包含以下

llm: 语言模型，进行文本生成
chain_type: 传入链类型

• stuff：将所有查询得到的文档组合成一个文档传入下一步。
• Map Reduce： 将所有块与问题一起传递给语言模型，获取回复，使用另一个语言模型调用将所有单独的回复总结成最终答案，它可以在任意数量的文档上运行。可以并行处理单个问题，同时也需要更多的调用。它将所有文档视为独立的
• Refine： 用于循环许多文档，实际上它是用迭代实现的，它建立在先前文档的答案之上，非常适合用于合并信息并随时间逐步构建答案，由于依赖于先前调用的结果，因此它通常需要更长的时间，并且基本上需要与Map Reduce一样多的调用
• Map Re-rank： 对每个文档进行单个语言模型调用，要求它返回一个分数，选择最高分，这依赖于语言模型知道分数应该是什么，需要告诉它，如果它与文档相关，则应该是高分，并在那里精细调整说明，可以批量处理它们相对较快，但是更加昂贵

实际代码示例：

from langchain.chains import RetrievalQA
from langchain.chat_models import ChatOpenAI
from langchain.document_loaders import CSVLoader
from langchain.vectorstores import DocArrayInMemorySearch
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains.question_answering import load_qa_chain# 1. 准备数据和向量存储
loader = CSVLoader('clothing_catalog.csv')
docs = loader.load()
embeddings = OpenAIEmbeddings()
db = DocArrayInMemorySearch.from_documents(docs, embeddings)# 初始化LLM
llm = ChatOpenAI(temperature=0)

1. Stuff Chain (直接组合)

# 适合处理少量文档，将所有文档直接组合
stuff_chain = RetrievalQA.from_chain_type(llm=llm,chain_type="stuff",retriever=db.as_retriever(search_kwargs={"k": 3}),verbose=True
)# 使用示例
query = "列出所有防晒衬衫的特点"
result = stuff_chain.run(query)
print(f"Stuff Chain 结果:\n{result}")

2. Map Reduce Chain (并行处理)

# 适合处理大量文档，将文档分块并行处理
map_reduce_chain = RetrievalQA.from_chain_type(llm=llm,chain_type="map_reduce",retriever=db.as_retriever(search_kwargs={"k": 5}),verbose=True
)# 使用示例 - 适合需要汇总的查询
query = "总结所有防晒服装的共同特点"
result = map_reduce_chain.run(query)
print(f"Map Reduce Chain 结果:\n{result}")# Map Reduce 处理过程
"""
1. Map阶段：每个文档单独处理
Doc1 -> LLM -> 结果1
Doc2 -> LLM -> 结果2
Doc3 -> LLM -> 结果32. Reduce阶段：合并所有结果
[结果1, 结果2, 结果3] -> LLM -> 最终结果
"""

3. Refine Chain (迭代优化)

# 适合需要渐进式构建答案的场景
refine_chain = RetrievalQA.from_chain_type(llm=llm,chain_type="refine",retriever=db.as_retriever(search_kwargs={"k": 4}),verbose=True
)# 使用示例 - 适合需要详细分析的查询
query = "分析防晒衬衫的材质和功能特点，并给出穿着建议"
result = refine_chain.run(query)
print(f"Refine Chain 结果:\n{result}")# Refine 处理过程
"""
初始答案 = LLM(Doc1)
改进答案1 = LLM(初始答案 + Doc2)
改进答案2 = LLM(改进答案1 + Doc3)
最终答案 = LLM(改进答案2 + Doc4)
"""

4. Map Rerank Chain (评分排序)

# 适合需要根据相关性排序的场景
map_rerank_chain = RetrievalQA.from_chain_type(llm=llm,chain_type="map_rerank",retriever=db.as_retriever(search_kwargs={"k": 5}),verbose=True
)# 使用示例 - 适合需要最相关答案的查询
query = "推荐最适合夏季穿着的防晒衬衫"
result = map_rerank_chain.run(query)
print(f"Map Rerank Chain 结果:\n{result}")# Map Rerank 处理过程
"""
1. 对每个文档生成答案和相关性分数
Doc1 -> LLM -> (答案1, 分数1)
Doc2 -> LLM -> (答案2, 分数2)
Doc3 -> LLM -> (答案3, 分数3)2. 选择分数最高的答案
返回 max(分数1, 分数2, 分数3) 对应的答案
"""

比较和使用建议：

def compare_chain_types(query):"""比较不同链类型的结果"""chains = {"stuff": stuff_chain,"map_reduce": map_reduce_chain,"refine": refine_chain,"map_rerank": map_rerank_chain}results = {}for name, chain in chains.items():print(f"\n使用 {name} chain 处理查询...")results[name] = chain.run(query)return results# 测试不同类型的查询
queries = {"简单查询": "有防晒衬衫吗？","汇总查询": "总结所有防晒服装的特点","详细分析": "分析不同防晒衬衫的优缺点","推荐查询": "推荐最适合户外活动的防晒衣物"
}for query_type, query in queries.items():print(f"\n\n处理 {query_type}: {query}")results = compare_chain_types(query)for chain_type, result in results.items():print(f"\n{chain_type} chain 结果:")print(result)

使用建议：

1. Stuff Chain:

   • 适用于文档数量少（ < 3-4个）
   • 需要快速响应
   • 文档内容简单

2. Map Reduce Chain:

   • 适用于大量文档
   • 需要综合信息
   • 可以并行处理
   • 适合统计和汇总类查询

3. Refine Chain:

   • 适用于需要深入分析的查询
   • 需要考虑上下文的连续性
   • 适合生成详细报告或分析

4. Map Rerank Chain:

   • 适用于需要精确匹配的查询
   • 需要按相关性排序的结果
   • 适合推荐类查询

这些链类型可以根据具体需求组合使用，比如先用Map Rerank找到最相关的文档，然后用Refine Chain生成详细分析。