大模型开发07：LangChain 大模型应用开发_langchain 模拟大模型返回

大模型开发07：LangChain 大模型应用开发

一、Model I/O 输入输出

任何语言模型应用程序的核心元素是什么？LangChain 提供了与任何语言模型交互的构建块。

• Prompts: 模板化、动态选择和管理模型输入
• Language models: 通过通用的接口访问大模型
• Output parsers: 大模型输出解析

模板化输入：Prompts

语言模型的提示是由用户提供的一组指令或输入，用于指导模型的响应，帮助它理解上下文并生成相关且一致的基于语言的输出，例如回答问题、完成句子或参与对话。LangChain 提供了几个类和函数来帮助构造和处理提示。

提示模板(Prompt templates)

参数化的提示模板，是一些预定义的片段用于生成模型所需的提示词，模板可能包括说明、少量示例以及适合给定任务的特定上下文和问题。LangChain 试图创建与模型无关的模板，以便在不同的语言模型之间轻松重用现有模板。

from langchain.prompts import PromptTemplate
prompt_template = PromptTemplate.from_template("Tell me a {adjective} joke about {content}.")
prompt_template.format(adjective="funny", content="chickens")
1
2
3

• 连接特征库以针对不同用户提供个性化服务

  class FeastPromptTemplate(StringPromptTemplate):
    def format(self, **kwargs) -> str:
    	driver_id = kwargs.pop("driver_id")
    	feature_vector = store.get_online_features(
    		features=[
    			"driver_hourly_stats:conv_rate",
    			"driver_hourly_stats:acc_rate",
    			"driver_hourly_stats:avg_daily_trips",
    		],
    		entity_rows=[{"driver_id": driver_id}],
    	).to_dict()
    	kwargs["conv_rate"] = feature_vector["conv_rate"][0]
    	kwargs["acc_rate"] = feature_vector["acc_rate"][0]
    	kwargs["avg_daily_trips"] = feature_vector["avg_daily_trips"][0]
    	return prompt.format(**kwargs)
  prompt_template = FeastPromptTemplate(input_variables=["driver_id"])
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16

• 自定义提示词模板，有两种不同的提示模板——字符串提示模板和聊天提示模板。字符串提示模板提供字符串格式的简单提示，而聊天提示模板生成更结构化的提示，用于聊天API。主要实现以下两方面：

  1. 一个 input_variables 属性，用于公开提示模板所期望的输入变量。
  2. 一个 format 方法，该方法接受与预期的 input_variables 对应的关键字参数，并返回格式化的提示符。
     如：

  class FunctionExplainerPromptTemplate(StringPromptTemplate, BaseModel):
    """函数解释提示词模板"""
  
    @validator("input_variables")
    def validate_input_variables(cls, v):
        """校验参数"""
        if len(v) != 1 or "function_name" not in v:
            raise ValueError("function_name must be the only input_variable.")
        return v
  
    def format(self, **kwargs) -> str:
        # 获取函数体源码
        source_code = get_source_code(kwargs["function_name"])
  
        # 组装提示词
        prompt = PROMPT.format(
            function_name=kwargs["function_name"].__name__, source_code=source_code
        )
        return prompt
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18
  19

• 少样本提示词模板，可以从一组示例或从一个示例选择器对象构造一个简短的提示模板。

  # 从一组示例创建少样本提示词
  examples = [{
    "question": "Who lived longer, Muhammad Ali or Alan Turing?",
    "answer": "..."
  },{...}]
  example_prompt = PromptTemplate(input_variables=["question", "answer"], template="Question: {question}\n{answer}")
  prompt = FewShotPromptTemplate(
        examples=examples,
        example_prompt=example_prompt,
        suffix="Question: {input}", # 添加最后问题模板
        input_variables=["input"]
    )
  # 调用
  print(prompt.format(input="Who was the father of Mary Ball Washington?"))
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14

• 局部初始化提示词模板

  prompt = PromptTemplate(template="{foo}{bar}", input_variables=["foo", "bar"])
  partial_prompt = prompt.partial(foo="foo");
  print(partial_prompt.format(bar="baz")) # 输出: foobaz
  # 方式2：使用 partial_variables
  prompt = PromptTemplate(template="{foo}{bar}", input_variables=["bar"], partial_variables={"foo": "foo"})
  print(prompt.format(bar="baz"))
  # 通过函数局部初始化
  prompt = PromptTemplate(template="Tell me a {adjective} joke about the day {date}",
    input_variables=["adjective", "date"]);
  partial_prompt = prompt.partial(date=_get_datetime)
  print(partial_prompt.format(adjective="funny"))
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11

• 组合模板

  input_prompts = [("introduction", introduction_prompt),
      ("example", example_prompt),
      ("start", start_prompt)]
  pipeline_prompt = PipelinePromptTemplate(final_prompt=full_prompt, pipeline_prompts=input_prompts)
  1
  2
  3
  4

• 序列化，支持从文件中加载模板，同时支持 JSON 和 YAML 格式，支持在一个文件中指定所有内容，或者在不同的文件中存储不同的组件 (模板、示例等) 并引用它们

  {
    "_type": "prompt",
    "input_variables": ["adjective", "content"],
    "template": "Tell me a {adjective} joke about {content}."
  }
  1
  2
  3
  4
  5

  prompt = load_prompt("simple_prompt.json")
  print(prompt.format(adjective="funny", content="chickens"))
  1
  2

• 字符串提示词模板流水线化

  prompt = (
    PromptTemplate.from_template("Tell me a joke about {topic}") # 第一个元素需要是提示词模板类型
    + ", make it funny"
    + "\n\nand in {language}"
  )
  1
  2
  3
  4
  5

示例选择器(Example selectors)

如果您有大量的示例，您可能需要动态选择在提示符中包含哪些示例。示例选择器是实现该功能的类。它需要定义的唯一方法是 select_examples 方法。它接受输入变量，然后返回一个示例列表。如何选择这些示例取决于每个具体实现。内置的选择器有长度选择器、相关性选择器、n-gram overlap 选择器、相似度选择器等

class BaseExampleSelector(ABC):
    """Interface for selecting examples to include in prompts."""

    @abstractmethod
    def select_examples(self, input_variables: Dict[str, str]) -> List[dict]:
    """Select which examples to use based on the inputs."""
1
2
3
4
5
6

语言模型：Language Models

LangChain 没有自己的语言模式，而是提供了一个标准接口与许多不同的大模型进行交互。主要为两类模型提供接口和集成:

1. 以文本字符串作为输入并返回文本字符串的语言模型
2. 但将聊天消息列表作为输入并返回聊天消息的聊天模型

LangChain 中的 LLM 指的是纯文本补全模型。它们的 api 接受字符串提示符作为输入，输出字符串补全。聊天模型通常由 LLMs 支持，但专门针对对话进行了调整。重要的是，它们的提供的 api 使用与纯文本补全模型不同的接口。它们将聊天消息列表作为输入，而不是单个字符串。通常，这些信息都标有角色 (通常是“system”，“ai” 和 “human”), 会返回一条聊天信息作为输出。GPT-4 和 Anthropic 的 Claude 都是作为聊天模型实现的。

为了能够兼容 LLMs 和聊天模型，两者都实现了 Base Language Model 接口。这包括常见的方法 “predict”，它接受一个字符串并返回一个字符串，以及“predict messages”，它接受一个消息并返回一个消息。如果正在使用特定的模型，建议使用特定于该模型的类的方法。

• 支持异步API: LangChain 通过利用 asyncio 库为 LLMs 提供异步支持。
• 支持自定义LLM包装器
• LangChain 提供了一个可用于测试的模拟 LLM 类。这允许您模拟对 LLM 的调用及输出。

responses = ["Action: ...", "Final Answer: ..."]
llm = FakeListLLM(responses=responses)
1
2

• LangChain 提供了一个可选的缓存层。可以通过减少 API 调用次数来节省资金，同时可以加快应用程序的响应速度
• 从本地文件加载 LLM 配置
• 支持流式响应，这意味着不必等待返回整个响应，而是可以在它可用时立即展示处理它
• 跟踪 Token 使用情况

规范化输出：Output Parsers

语言模型输出文本。但很多时候，你可能想要得到更多结构化的信息，而不仅仅是短信回复。这就是输出解析器的用武之地。输出解析器需要实现的方法:

1. “返回格式化指令”: 一个返回字符串的方法，其中包含语言模型的输出应该如何格式化的指令。
2. “解析方法”: 一个接受字符串(假设是语言模型的响应)并将其解析为某种结构的方法。
3. “Parse with prompt”(可选): 一个方法，它接受一个字符串(假设是来自语言模型的响应)和一个提示(假设是生成这样一个响应的提示)，并将其解析成某种结构。提示主要是在 OutputParser 想要以某种方式重试或修复输出，并且需要来自提示的信息时提供的。
   LangChain 内置的解析器：列表解析器、日期解析器、枚举解析器、自修复解析器、Json 解析器、结构化解析器、XML 解析器

output_parser = CommaSeparatedListOutputParser()

format_instructions = output_parser.get_format_instructions()
prompt = PromptTemplate(
    template="List five {subject}.\n{format_instructions}",
    input_variables=["subject"],
    partial_variables={"format_instructions": format_instructions}
)

model = OpenAI(temperature=0)

_input = prompt.format(subject="ice cream flavors")
output = model(_input)

output_parser.parse(output)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

二、Retrieval 检索

许多 LLM 应用需要使用不属于模型训练集的用户特定数据。实现这一目标的主要方法是通过检索增强生成(RAG)。在这个过程中，外部数据被检索，然后在执行生成步骤时作为补充内容传递给 LLM。LangChain 为 RAG 应用程序提供了从简单到复杂的所有构建块。

• 文档加载：LangChain 提供了超过100种不同的文档加载器
• 文档转换：检索的核心是获取文档中相关的部分。这涉及到几个转换步骤，以便为检索做好最好的准备。主要方法之一是将大文档分割(或分块)为较小的块。LangChain 有许多内置的文档转换器，可以方便地拆分、组合、过滤和操作文档。
• 文本嵌入：检索的另一个关键部分是为文档创建嵌入。嵌入捕获文本的语义含义，允许您快速有效地找到相似的其他文本片段。LangChain 提供了超过 25 种不同的嵌入提供商和方法的集成，从开源到专有 API，允许您选择最适合的一个。并且提供了一个标准接口，可以轻松地在模型之间进行切换。
• 向量存储：随着嵌入的兴起，出现了对数据库支持这些嵌入的高效存储和搜索的需求。LangChain 提供了超过 50 种不同向量数据库的集成。
• 检索器：LangChain 支持许多不同的检索算法。检索器接受字符串查询作为输入，并返回 Document 类型列表作为输出。
• 索引：索引 API 允许您将来自任何源的文档加载到向量数据库中并保持同步。

文档加载器

最简单的加载器将文件作为文本读入，并将其全部放入一个文档中。

from langchain.document_loaders import TextLoader

loader = TextLoader("./index.md")
loader.load()
1
2
3
4

• CSV加载器：加载CSV数据，每行对应一个 Document 数据。
• 批量加载目录文件：默认情况下，它使用 UnstructuredLoader 类加载文件。但是可以很容易地更改加载器的类型。

loader = DirectoryLoader('../', glob="**/*.md", loader_cls=TextLoader)
docs = loader.load()
1
2

• HTML 文档加载器，可以使用 BSHTMLLoader 利用 BeautifulSoup4 加载 HTML 文档。

loader = UnstructuredHTMLLoader("example_data/fake-content.html")
loader = BSHTMLLoader("example_data/fake-content.html")
1
2

• JSON 加载器：JSONLoade r使用 jq 模式来解析 JSON 文件。它使用 jq python 包。
• markdown 加载器：loader = UnstructuredMarkdownLoader(markdown_path)
• Pdf 加载器：使用 pypdf 将 PDF 加载到 Document 数组中，其中每个文档包含页面内容和带有页码的元数据。

loader = PyPDFLoader("example_data/layout-parser-paper.pdf")
pages = loader.load_and_split()
1
2

文档转换器

文本切片，当您想要处理长文本时，有必要将该文本分割成块。虽然这听起来很简单，但这里有很多潜在的复杂性。理想情况下，您希望将语义相关的文本片段保持在一起。文本分割器的工作方式如下:

1. 将文本分成语义上有意义的小块(拆分规则：通常是句子)。
2. 将这些小块组合成一个更大的块，直到达到一定的大小(判断块大小：由某个函数测量)。
3. 一旦你达到这个大小，让这个块成为它自己的文本块，然后创建一个有一些重叠的新文本块(以保持块之间的上下文)。

默认推荐的文本分割器是 RecursiveCharacterTextSplitter。这个文本分割器接受一个字符列表。它尝试基于分割第一个字符创建块，但如果任何块太大，它就会移动到下一个字符，以此类推。

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size = 100, # Set a really small chunk size, just to show.
    chunk_overlap  = 20,
    length_function = len,
    add_start_index = True,
)
1
2
3
4
5
6

• HTMLHeaderTextSplitter 是一个“结构感知”的分块器，它在元素级别拆分文本，并为每个与任何给定块“相关”的头添加元数据。它可以一个元素一个元素地返回块，或者将具有相同元数据的元素组合起来，其目的是: (a)在语义上保持相关文本的分组(或多或少)，以及(b)保留在文档结构中编码的上下文丰富的信息。它可以与其他文本分割器一起使用，作为分块管道的一部分。

headers_to_split_on = [
    ("h1", "Header 1"),
    ("h2", "Header 2"),
    ("h3", "Header 3"),
]

html_splitter = HTMLHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
html_header_splits = html_splitter.split_text(html_string)

# 使用其他文本分割器进一步处理
chunk_size = 500
chunk_overlap = 30
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=chunk_size, chunk_overlap=chunk_overlap
)
splits = text_splitter.split_documents(html_header_splits)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

• CharacterTextSplitter 按字符分割，这是最简单的方法。它基于字符(默认情况下为“\n\n”)进行分割，并根据字符数测量块长度。
• CodeTextSplitter 允许您拆分支持多种语言的代码。
• 语言模型有一个令牌限制不应超限制。因此，当您将文本分割成块时，需要计算 token 的数量。

text_splitter = CharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=100, chunk_overlap=0)
texts = text_splitter.split_text(state_of_the_union)
1
2
3

• MarkdownHeaderTextSplitter 按一组指定的标头拆分一个 markdown 文件。

headers_to_split_on = [
    ("#", "Header 1"),
    ("##", "Header 2"),
    ("###", "Header 3"),
]

markdown_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
md_header_splits = markdown_splitter.split_text(markdown_document)
1
2
3
4
5
6
7
8

文本嵌入

Embeddings 类是一个设计用于与文本嵌入模型交互的类。有很多嵌入模型提供程序(OpenAI, Cohere, Hugging Face 等)，这个类的目的是为它们提供一个标准接口。嵌入创建一段文本的向量表示，这意味着我们可以考虑向量空间中的文本，并进行语义搜索，在向量空间中寻找最相似的文本片段。LangChain 中的 Embeddings 类提供了两个方法: 一个用于嵌入文档，另一个用于嵌入查询。

from langchain.embeddings import OpenAIEmbeddings
embeddings_model = OpenAIEmbeddings()
embeddings = embeddings_model.embed_documents(["Hi there!", "Oh, hello!", ...])

# 嵌入查询
embedded_query = embeddings_model.embed_query("What was the name mentioned in the conversation?")
1
2
3
4
5
6

向量数据库

向量数据库负责存储嵌入的数据并为您执行向量搜索。

from langchain.document_loaders import TextLoader
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.text_splitter import CharacterTextSplitter
from langchain.vectorstores import Chroma

# 加载文档, 切分成文本块, 嵌入为向量存储到数据库
raw_documents = TextLoader('../../../state_of_the_union.txt').load()
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
documents = text_splitter.split_documents(raw_documents)
db = Chroma.from_documents(documents, OpenAIEmbeddings())

# 相似搜索
query = "What did the president say about Ketanji Brown Jackson"
docs = db.similarity_search(query)
print(docs[0].page_content)

# 通过向量搜索
embedding_vector = OpenAIEmbeddings().embed_query(query)
docs = db.similarity_search_by_vector(embedding_vector)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

检索器

检索器是一个接口，它在给定非结构化查询时返回文档。它比向量数据库更通用。检索器不需要能存储文档，只需要能够返回(或检索)它们。向量数据库可以用作检索器的核心，但也有其他类型的检索器。

• MultiQueryRetriever 通过使用 LLM 从不同的角度为给定的用户输入查询生成多个查询，从而使提示调优过程自动化。能够克服基于距离的检索的一些限制，并获得更丰富的结果集。

question = "What are the approaches to Task Decomposition?"
llm = ChatOpenAI(temperature=0)
retriever_from_llm = MultiQueryRetriever.from_llm(
    retriever=vectordb.as_retriever(), llm=llm
)
unique_docs = retriever_from_llm.get_relevant_documents(query=question)
1
2
3
4
5
6

• 上下文压缩
  检索的一个挑战是，给定一个示例问题，我们的检索器返回一两个相关的文档和一些不相关的文档。即使是相关的文档也有很多不相关的信息。这意味着与查询最相关的信息可能隐藏在包含大量无关文本的文档中。通过应用程序传递完整的文档可能会导致更昂贵的 LLM 调用和更差的响应。
  上下文压缩就是为了解决这个问题。其思想很简单: 与其按原样立即返回检索到的文档，不如使用给定查询的上下文压缩它们，以便只返回相关信息。这里的“压缩”既指压缩单个文档的内容，也指过滤掉整个文档。

from langchain.llms import OpenAI
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

llm = OpenAI(temperature=0)
# LLMChainExtractor，它将遍历最初返回的文档，并仅从每个文档中提取与查询相关的内容。
compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(base_compressor=compressor, base_retriever=retriever)

compressed_docs = compression_retriever.get_relevant_documents("What did the president say...")
pretty_print_docs(compressed_docs)
1
2
3
4
5
6
7
8
9
10
11

• EnsembleRetriever 将检索器列表作为输入，并汇总它们的 get_relevance _documents() 方法的结果，并基于互反秩融合算法对结果重新排序。最常见的模式是将稀疏检索器(如BM25)与密集检索器(如嵌入相似度)结合起来，因为它们的优势是互补的。它也被称为“混合搜索”。稀疏检索器擅长根据关键词找到相关文档，而密集检索器擅长根据语义相似度找到相关文档。

# initialize the bm25 retriever and faiss retriever
bm25_retriever = BM25Retriever.from_texts(doc_list)
bm25_retriever.k = 2

embedding = OpenAIEmbeddings()
faiss_vectorstore = FAISS.from_texts(doc_list, embedding)
faiss_retriever = faiss_vectorstore.as_retriever(search_kwargs={"k": 2})

# initialize the ensemble retriever
ensemble_retriever = EnsembleRetriever(retrievers=[bm25_retriever, faiss_retriever], weights=[0.5, 0.5])
1
2
3
4
5
6
7
8
9
10

• MultiVector Retriever 多重向量检索器，为每个文档创建多个向量。为每个文档创建多个矢量的方法包括:
  1. 更小的块: 将文档分割成更小的块，并嵌入它们。
  2. 摘要: 为每个文档创建一个摘要，将其与文档一起嵌入(或代替)。
  3. 提问: 创建每个文档都适合回答的问题，将问题与文档一起嵌入(或代替)。
• 父文档检索器，在分割文档进行检索时，经常会有相互冲突的需求: 既希望使用较小的文档，以便嵌入可以最准确地反映其含义。如果过长，嵌入内容就会失去意义。又希望拥有足够长的文档，以便保留每个块的上下文ParentDocumentRetriever 通过分割和存储小块数据来实现这种平衡。在检索期间，它首先获取小块，然后查找这些块的父 id 并返回较大的文档。“父文档”指的是小块源自的文档。这可以是整个原始文档，也可以是更大的块。

child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)
# The vectorstore to use to index the child chunks
vectorstore = Chroma(
    collection_name="full_documents",
    embedding_function=OpenAIEmbeddings()
)
# The storage layer for the parent documents
store = InMemoryStore()
retriever = ParentDocumentRetriever(
    vectorstore=vectorstore, 
    docstore=store, 
    child_splitter=child_splitter,
)
retriever.add_documents(docs, ids=None)
# 向量数据库中可检索小文本块
sub_docs = vectorstore.similarity_search("justice breyer")
# 返回较小块所在的文档
retrieved_docs = retriever.get_relevant_documents("justice breyer")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

• 自查询检索器，顾名思义，是具有查询自身能力的检索器。具体来说，给定任何自然语言查询，检索器使用一个构造查询的 LLM 链来编写结构化查询，然后将该结构化查询应用于其底层 VectorStore。这使得检索器不仅可以使用用户输入查询与向量数据进行语义相似性比较，还可以从用户查询中提取应用于文档元数据上的过滤器，并执行这些过滤器。
• 时间加权向量存储检索器，这个检索器结合了语义相似性和时间衰减。评分算法为: semantic_similarity + (1.0 - decay_rate) ^ hours_passed, hours_passed 指的是自上次访问检索器中的对象以来经过的小时数，而不是自创建对象以来经过的小时数。这意味着频繁访问的对象会保持“新鲜”。
• WebResearchRetriever，制定一组相关谷歌搜索，执行每一个搜索，加载所有结果，然后在合并页面内容上嵌入并执行相似度搜索。

索引

索引 API 允许您将来自任何源的文档加载到向量数据库中并保持同步。具体来说，它有助于:

• 避免将重复的内容写入向量数据库
• 避免重写未更改的内容
• 避免在未更改的内容上重新计算嵌入

LangChain 索引使用记录管理器 (RecordManager) 来跟踪写入向量数据库的文档。当索引内容时，为每个文档计算哈希值，并将以下信息存储在记录管理器中:

• 文档散列(页面内容和元数据的散列)
• 写入时间
• 源id——每个文档应该在其元数据中包含信息，以便我们确定该文档的最终来源

collection_name = "test_index"
embedding = OpenAIEmbeddings()
vectorstore = ElasticsearchStore(es_url="http://localhost:9200", index_name="test_index", embedding=embedding)
namespace = f"elasticsearch/{collection_name}"
record_manager = SQLRecordManager(namespace, db_url="sqlite:///record_manager_cache.sql")
# 在使用记录管理器之前创建 schema。
record_manager.create_schema()
# 待创建索引的测试文档
doc1 = Document(page_content="kitty", metadata={"source": "kitty.txt"})
doc2 = Document(page_content="doggy", metadata={"source": "doggy.txt"})

def _clear():
    """清理"""
    index([], record_manager, vectorstore, cleanup="full", source_id_key="source")
# 建立索引
index(
    [doc1, doc1, doc1, doc1, doc1],
    record_manager,
    vectorstore,
    cleanup=None,
    source_id_key="source",
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

三、Chains 链

对于简单的应用程序，单独使用 LLM 是可以的，但是更复杂的应用程序需要将 LLMs 链接起来——要么相互链接，要么与其他组件链接。

LangChain 提供了两个用于“链接”组件的高级框架。旧方法是使用 Chain 接口。更新的推荐方法是使用 LangChain 表达式语言(LCEL)。但是有许多有用的、内置的链，将继续支持，Chain 本身也可以在LCEL中使用，因此两者并不相互排斥。

LCEL最明显的部分是它为组合提供了直观和可读的语法。但更重要的是，它还提供了以下支持: 流, 异步调用, 批处理, 并行化, 重试, 回退, 跟踪和更多。

一个简单而常见的例子是将提示符、模型和输出解析器组合在一起:

from langchain.chat_models import ChatAnthropic
from langchain.prompts import ChatPromptTemplate
from langchain.schema import StrOutputParser

model = ChatAnthropic()
prompt = ChatPromptTemplate.from_messages([
    ("system", "You're a very knowledgeable historian who provides ..."),
    ("human", "{question}"),
])

runnable = prompt | model | StrOutputParser()
for chunk in runnable.stream({"question": "How did Mansa Musa accumulate his wealth?"}):
    print(chunk, end="", flush=True)
1
2
3
4
5
6
7
8
9
10
11
12
13

• Chain 接口类，使用内置的 LLMChain 实现上面的 LCEL 功能 :

from langchain.chains import LLMChain

chain = LLMChain(llm=model, prompt=prompt, output_parser=StrOutputParser())
chain.run(question="How did Mansa Musa accumulate his wealth?")
1
2
3
4

LLMChain

LLMChain 是一个简单的链，语言模型增加功能。它在整个 LangChain 中广泛使用，包括在其他链和代理中。它由 PromptTemplate 和语言模型组成。它使用输入参数或使用内存键值对)格式化提示词模板，将格式化后的字符串传递给 LLM，并返回 LLM 输出。

除了所有 Chain 对象都有的 __call__ 和 run 方法外，它还提供了其他几个方法：

• apply, 允许你根据输入列表运行链
• generate, 类似于apply，它返回 LLMResult 而不是字符串。通常包含 token 用量和结束原因
• predict, 类似于 run 方法，不同之处在于输入键被指定为关键字参数而不是 Python 字典
• 使用 predict_and_parse 和 apply_and_parse 可以将输出解析器作用于 LLM 输出

RouterChain

路由链，它可以动态地根据输入选择下一个链。

MultiPromptChain 创建一个问答路由链，该链选择与给定问题最相关的提示，然后使用该提示回答问题。

physics_template = """You are a very smart physics professor. ...Here is a question:
{input}"""
math_template = """You are a very good mathematician. ...Here is a question:
{input}"""
prompt_infos = [{
        "name": "physics",
        "description": "Good for answering questions about physics",
        "prompt_template": physics_template,
    }, {
        "name": "math",
        "description": "Good for answering math questions",
        "prompt_template": math_template,
    }]
# 使用 openai 的模型
llm = OpenAI()
# 每个提示词模板生成一个 destination_chain
destination_chains = {}
for p_info in prompt_infos:
    name = p_info["name"]
    prompt_template = p_info["prompt_template"]
    prompt = PromptTemplate(template=prompt_template, input_variables=["input"])
    chain = LLMChain(llm=llm, prompt=prompt)
    destination_chains[name] = chain
default_chain = ConversationChain(llm=llm, output_key="text")

# 使用内置模板 MULTI_PROMPT_ROUTER_TEMPLATE 创建路由链
destinations = [f"{p['name']}: {p['description']}" for p in prompt_infos]
destinations_str = "\n".join(destinations)
router_template = MULTI_PROMPT_ROUTER_TEMPLATE.format(destinations=destinations_str)
router_prompt = PromptTemplate(
    template=router_template,
    input_variables=["input"],
    output_parser=RouterOutputParser(),
)
router_chain = LLMRouterChain.from_llm(llm, router_prompt)

# 创建多模板路由链
chain = MultiPromptChain(
    router_chain=router_chain,
    destination_chains=destination_chains,
    default_chain=default_chain,
    verbose=True,
)
print(chain.run("What is black body radiation?"))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

顺序链

对语言模型进行一连串调用，从一个调用中获取输出并将其作为另一个调用的输入。顺序链允许您连接多个链，并将它们组合成执行某些特定场景的管道。有两种类型的顺序链:

• SimpleSequentialChain: 顺序链的最简单形式，其中每个步骤都有一个单一的输入/输出，一个步骤的输出是下一个步骤的输入。

# 传递单字符串作为参数，所有步骤的输出也是单个字符串
overall_chain = SimpleSequentialChain(chains=[synopsis_chain, review_chain], verbose=True)
review = overall_chain.run("Tragedy at sunset on the beach")
1
2
3

• SequentialChain: 一种更通用的顺序链形式，允许多个输入/输出。

overall_chain = SequentialChain(
    chains=[synopsis_chain, review_chain],
    input_variables=["era", "title"],
    # Here we return multiple variables
    output_variables=["synopsis", "review"],
    verbose=True)
1
2
3
4
5
6

转换链

例如，将文本过滤到仅前 3 段，然后将其传递到 LLMChain 中以总结这些文本

def transform_func(inputs: dict) -> dict:
    text = inputs["text"]
    shortened_text = "\n\n".join(text.split("\n\n")[:3])
    return {"output_text": shortened_text}
transform_chain = TransformChain(
    input_variables=["text"], output_variables=["output_text"], transform=transform_func
)
template = """Summarize this text:
{output_text}
Summary:"""
prompt = PromptTemplate(input_variables=["output_text"], template=template)
llm_chain = LLMChain(llm=OpenAI(), prompt=prompt)
sequential_chain = SimpleSequentialChain(chains=[transform_chain, llm_chain])
sequential_chain.run("...")
1
2
3
4
5
6
7
8
9
10
11
12
13
14

文档链

• Stuff 链，适合文档较小且大多数调用只传递少量文档的应用。
• Refine 链，通过循环遍历输入文档并迭代更新其答案来构造响应。对于每个文档，它将所有非文档输入、当前文档和最新的中间答案传递给 LLM 链，以获得新答案。
• Mapreduce 链，首先对每个文档单独应用 LLM 链 ( map 步骤)，将链输出作为新文档。然后传递给单独的组合文档链以获得单个输出 ( Reduce 步骤)。它可以选择首先压缩或折叠映射的文档，以确保它们适合组合文档链(通常会将它们传递给 LLM)。如有必要，此压缩步骤将递归执行。
• Map re-rank 链在每个文档上运行一个初始提示，它不仅尝试完成任务，而且还对其答案的确定程度进行评分, 最终返回得分最高的响应。

四、Memory 记忆

大多数 LLM 应用程序都有会话功能。对话的一个重要组成部分是能够引用对话之前介绍的信息。至少，会话系统应该能够直接访问过去的某些对话消息。更复杂的系统需要有一个不断更新的世界模型，使它有能力做一些其他事情，比如维护实体及其关系的信息。
把这种储存过去互动信息的能力称为“记忆”。LangChain 提供了许多用于向系统添加记忆的实用工具。这些工具可以单独使用，也可以无缝地添加到一个链中。

1. 在接收到初始用户输入之后，但在执行核心逻辑之前，链将从其内存系统中读取并增加用户输入。
2. 在执行核心逻辑之后，但在返回答案之前，链将把当前运行的输入和输出写入内存，以便在以后的运行中引用它们。

记忆系统设计

1. 存储: 聊天消息列表

   任何记忆的底层都是所有聊天交互的历史记录。即使不直接使用它们，也需要以某种形式存储它们。LangChain 记忆模块的关键部分之一是用于存储这些聊天消息的一系列集成，从内存列表到持久数据库。

2. 查询: 聊天消息的数据结构和算法
   建立在聊天消息之上的数据结构和算法，为这些消息提供最有用的视图。一个非常简单的记忆系统每次运行时可能只返回最近的消息。稍微复杂一点的可能会返回过去多条消息的摘要。更复杂的系统可能会从存储的消息中提取实体，并且只返回引用的实体的信息。

有许多不同类型的记忆工具，每一种都有自己的参数及返回类型，适用不同的场景。

• ConversationBufferMemory 用于存储消息，然后在变量中提取全部消息。
• ConversationBufferWindowMemory 保存一段时间内对话交互的列表。它只用最后 K 个记录。这对于保持最近交互的滑动窗口非常有用，这样缓冲区就不会变得太大。
• ConversationEntityMemory 用于记忆对话中特定实体的信息。它提取实体的信息(使用 LLM )，并随着时间的推移 (也使用LLM) 建立关于该实体的知识。
• ConversationKGMemory 这种类型的记忆使用知识图谱来重建记忆。
• ConversationSummaryMemory，对话摘要记忆会随着时间的推移对谈话进行总结。这对于浓缩对话中的信息很有用，可以将迄今为止的对话摘要注入提示/链中。
• VectorStoreRetrieverMemory 将内存存储在向量数据库中，并在每次调用时查询 top-K 最“突出”的文档（不显式地跟踪交互的顺序），“文档”是以前的对话片段，用于在对话中告知相关信息。
• CombinedMemory，在同一链中使用多个内存类。

llm = OpenAI(temperature=0)
# 其中模板参数 chat_history 来自记忆模块
template = """You are a nice chatbot having a conversation with a human.
Previous conversation:
{chat_history}
New human question: {question}
Response:"""
prompt = PromptTemplate.from_template(template)

# 设置 `memory_key` 与模板一致
memory = ConversationBufferMemory(memory_key="chat_history")
conversation = LLMChain(llm=llm, prompt=prompt, verbose=True, memory=memory)

# 只需要设置 `question` 变量，`chat_history` 从记忆模块提取
conversation({"question": "hi"})
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

自定义记忆类

在 LangChain 中有一些预定义的记忆类型，还可以创建适合自己的应用程序的记忆类。下例中，将演示如何编写一个自定义内存类，它使用 spaCy 提取实体，并将有关实体的信息保存在一个简单的散列表中。然后，在对话期间，我们将根据输入提取实体，并将信息放入上下文中。

import spacy
# 加载英文模型, 以进行分词
nlp = spacy.load("en_core_web_lg")

class SpacyEntityMemory(BaseMemory, BaseModel):
    """自定义记忆类，用于存储实体信息"""

    # 存储实体信息的字典
    entities: dict = {}
    # 定义键以将有关实体的信息传递到提示词模板
    memory_key: str = "entities"

    def clear(self):
        self.entities = {}

    @property
    def memory_variables(self) -> List[str]:
        return [self.memory_key]

    def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, str]:
        """加载内存实体信息"""
        # 通过 spaCy 从输入文本获取实体
        doc = nlp(inputs[list(inputs.keys())[0]])
        # 提取已知的实体信息
        entities = [
            self.entities[str(ent)] for ent in doc.ents if str(ent) in self.entities
        ]
        # 返回拼接的实体信息
        return {self.memory_key: "\n".join(entities)}

    def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None:
        """存储记忆到记忆缓存"""
        text = inputs[list(inputs.keys())[0]]
        doc = nlp(text)
        # 针对每个实体，存储信息到字典中
        for ent in doc.ents:
            ent_str = str(ent)
            if ent_str in self.entities: # 已存在则拼接
                self.entities[ent_str] += f"\n{text}"
            else:
                self.entities[ent_str] = text
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

五、Agents 代理

代理的核心思想是使用 LLM 来选择要采取的一系列行动。在链中，一系列动作是硬编码的(用代码)。在代理中，语言模型被用作推理引擎，以确定以何种顺序采取哪些操作。

• AgentAction: 这个数据类表示代理应该执行的操作，它有一个 tool 属性 (被调用的工具的名称) 和一个tool_input属性(工具的输入)
• AgentFinish: 这个数据类表示代理已经完成并应该返回给用户。它有一个 return_values 参数，是一个要返回的字典。通常只有一个键-output-是一个字符串，通常只是返回这个键。
• intermediate_steps: 表示先前的代理操作和相应输出。这些信息传递给未来的迭代是很重要的，这样代理就知道它已经完成了什么工作。它的类型是 List[Tuple[AgentAction, Any]]。请注意，观测值目前保留为 Any 类型，以获得最大的灵活性。在实践中通常是一个字符串。

下面介绍该模块的几个主要组件：

1. Agent

这是负责决定下一步采取什么步骤的 Chain，基于语言模型和提示词来实现。这个链的输入是:

• 可用工具列表
• 用户输入
• 任何先前执行的步骤 (intermediate_steps)

然后，该链返回要采取的下一个操作或要发送给用户的最终响应 (AgentAction 或 AgentFinish)。不同的代理具有不同的推理提示词风格、输入编码方式和输出解析方式。

2. Tools

工具是 Agent 调用的函数，一是让代理可以访问正确的工具，二是以对代理最有帮助的方式描述工具。LangChain提供了一套广泛的工具直接使用，同时也可以很容易地定义自己的工具(包括自定义描述)。

在构造代理时，需要为它提供一个工具列表。除了实际调用的函数外，工具由以下几个组件组成:

• Name(str), 必需，并且在给 Agent 的一组工具中名称必须是唯一的
• Description (str)是可选的，但推荐使用，因为代理使用它来确定使用哪个工具
• return_direct (bool)，默认为 False
• args_schema (Pydantic BaseModel) 是可选的，但推荐使用，它可以用来提供更多信息(例如示例)或对参数进行验证。

为了更容易定义自定义工具，提供了 @tool 装饰器。这个装饰器可以用来从一个简单的函数快速创建一个工具。装饰器默认使用函数名作为工具名，但可以通过传递一个字符串作为第一个参数来覆盖这一点。此外，装饰器将使用函数的注释作为工具的描述。

from langchain.tools import tool
@tool("search", return_direct=True, args_schema=SearchInput)
def search_api(query: str) -> str:
    """Searches the API for the query."""
    return f"Results for query {query}"
1
2
3
4
5

3. Toolkits

通常，代理可以访问的工具集比单个工具更重要。为此，LangChain提供了工具包的概念——完成特定目标所需的工具包,一个工具包中通常有3-5个工具。

4. AgentExecutor

代理执行器是代理的运行时, 调用代理并执行它选择的操作。为您处理一些复杂的问题: 代理选择了不存在的工具，工具出错的情况，代理产生的输出无法解析为工具调用的情况，所有级别(代理决策、工具调用)的日志记录和可观察性。

基于 LCEL 的 Agent 开发

构造一个可以访问自定义工具的自定义代理。在这个例子中展示如何使用 LCEL(LangChain Expression Language) 从头构建这个代理，这个代理基于 OpenAI 的函数调用来实现，给 Agent 的工具是一个计算单词长度的工具，并展示如何结合记忆模块。

from langchain.agents import tool
from langchain.chat_models import ChatOpenAI

# 基于 openai 的对话模型
llm = ChatOpenAI(temperature=0)

# 定义一个计算字符串长度的工具
@tool
def get_word_length(word: str) -> int:
    """Returns the length of a word."""
    return len(word)

tools = [get_word_length]
1
2
3
4
5
6
7
8
9
10
11
12
13

创建提示词模板。因为 OpenAI 的函数调用针对工具使用进行了微调，所以我们几乎不需要任何关于如何推理或如何输出格式的说明。只有两个输入变量: input(用户问题) 和 agent_scratchpad (之前采取的步骤)。

from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.schema.messages import HumanMessage, AIMessage
from langchain.agents.output_parsers import OpenAIFunctionsAgentOutputParser

MEMORY_KEY = "chat_history"
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are very powerful assistant, but bad at calculating lengths of words."),
    MessagesPlaceholder(variable_name=MEMORY_KEY),
    ("user", "{input}"),
    MessagesPlaceholder(variable_name="agent_scratchpad"),
])
# 将工具绑定到到 LLM
from langchain.tools.render import format_tool_to_openai_function
llm_with_tools = llm.bind(
    functions=[format_tool_to_openai_function(t) for t in tools]
)


# 组装 Agent
chat_history = []
agent = {
	# 模板参数 'input'
    "input": lambda x: x["input"],
    # 表示先前的代理操作和相应输出
    "agent_scratchpad": lambda x: format_to_openai_functions(x['intermediate_steps'])
    "chat_history": lambda x: x["chat_history"]
    # Parser 将输出转换为 AgentAction 或 AgentFinish。
} | prompt | llm_with_tools | OpenAIFunctionsAgentOutputParser()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

创建好 Agent 后的调用流程：

intermediate_steps = [] # 中间步骤
while True:
    output = agent.invoke({
        "input": "how many letters in the word educa?",
        "intermediate_steps": intermediate_steps
    })
    if isinstance(output, AgentFinish): # 完成
        final_result = output.return_values["output"]
        break
    else: # 动作执行函数调用
        print(output.tool, output.tool_input)
        tool = {
            "get_word_length": get_word_length
        }[output.tool]
        observation = tool.run(output.tool_input)
        intermediate_steps.append((output, observation))
print(final_result)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

可以使用 AgentExecutor 类简化上面的调用流程。它将上述所有功能捆绑在一起，并添加了错误处理、提前终止、跟踪和其他改进，从而减少了需要编写的保护逻辑。

from langchain.agents import AgentExecutor
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
input1 = "how many letters in the word educa?"
result = agent_executor.invoke({"input": input1})
chat_history.append(HumanMessage(content=input1)) # 保存该轮对话记忆
chat_history.append(AIMessage(content=result['output']))
# 第二轮对话
agent_executor.invoke({"input": "is that a real word?", "chat_history": chat_history})
1
2
3
4
5
6
7
8

六、Callbacks 回调

LangChain 提供了一个回调系统，允许添加钩子到 LLM 应用程序的各个阶段。这对于日志记录、监视、流处理和其他任务非常有用。

Callback handlers 回调处理器

Callback Handlers 是实现了 CallbackHandler 接口的对象，该接口为每个可以订阅的事件提供一个方法。当事件被触发时，CallbackManager 将调用对应的方法。需要实现下面的一个或多个方法：

class BaseCallbackHandler:
    """Base callback handler that can be used to handle callbacks from langchain."""
    def on_llm_start()
    def on_chat_model_start()
    def on_llm_new_token()
    def on_llm_end()
    def on_llm_error()
    def on_chain_start()
    def on_chain_end()
    def on_chain_error()
    def on_tool_start()
    def on_tool_end()
    def on_tool_error()
    def on_text() -> Any:
    def on_agent_action()
    def on_agent_finish()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

自定义回调处理器：

class MyCustomHandler(BaseCallbackHandler):
    def on_llm_new_token(self, token: str, **kwargs) -> None:
        print(f"My custom handler, token: {token}")
1
2
3

LangChain 提供了一些内置处理器(langchain/callbacks) 。最基本的一个处理器是StdOutCallbackHandler，它将所有事件记录到标准输出。当 verbose 设置为 true 时，即使没有显式传递，也会调用 StdOutCallbackHandler。callbacks 参数在 API 中的大多数对象(链、模型、工具、代理等)上都是可用的。

handler = StdOutCallbackHandler()
llm = OpenAI()
prompt = PromptTemplate.from_template("1 + {number} = ")

# 构造函数设置回调: 在初始化链时显式设置 StdOutCallbackHandler
chain = LLMChain(llm=llm, prompt=prompt, callbacks=[handler])
chain.run(number=2)

# 使用 verbose 标志来达到相同的效果，适用于记录所有请求
chain = LLMChain(llm=llm, prompt=prompt, verbose=True)
chain.run(number=2)

# 使用请求回调来实现相同的效果, 想要将单个特定的 websocket 连接请求流式输出
chain = LLMChain(llm=llm, prompt=prompt)
chain.run(number=2, callbacks=[handler])
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15