本电子书开源，欢迎 star 🌟，关注《LLM 应用开发实践笔记》

我的新书《LangChain编程从入门到实践》 已经开售！推荐正在学习AI应用开发的朋友购买阅读！ LangChain编程从入门到实践

LangChain模块解读

LangChain 分为 6 个模块，分别是对（大语言）模型输入输出的管理、外部数据接入、链的概念、（上下文记忆）存储管理、智能代理以及回调系统，通过文档的组织结构，你可以清晰了解到 LangChain的侧重点，在大语言模型开发生态中对自己的定位。下面我将对各个模型逐个进行解读。

Model I/O

这部分包括对大语言模型输入输出的管理，输入环节的提示词管理（包含模板化提示词和提示词动态选择等），处理环节的语言模型（包括所有LLMs的通用接口，以及常用的LLMs工具；Chat模型是一种与LLMs不同的API，用来处理消息），输出环节包括从模型输出中提取信息。

提示词管理

  • 提示模板 动态提示词=提示模板+变量，通过引入给提示词引入变量的方式，一方面保证了灵活性，一方面又能保证Prompt内容结构达到最佳

      from langchain import PromptTemplate
  no_input_prompt = PromptTemplate(input_variables=[], template="Tell me a joke.")
  no_input_prompt.format()

  one_input_prompt = PromptTemplate(input_variables=["adjective"], template="Tell me a {adjective} joke.")
  # "Tell me a funny chickens."
  one_input_prompt.format(adjective="funny")

  multiple_input_prompt = PromptTemplate(
      input_variables=["adjective", "content"], 
      template="Tell me a {adjective} joke about {content}."
  )
  # "Tell me a funny joke about chickens."
  multiple_input_prompt.format(adjective="funny", content="chickens")

  • 聊天提示模板 聊天场景中，消息可以与AI、人类或系统角色相关联，模型应该更加密切地遵循系统聊天消息的指示。这个是对 OpenAI gpt-3.5-tubor API中role字段（role 的属性用于显式定义角色，其中 system 用于系统预设，比如”你是一个翻译家“，“你是一个写作助手”，user 表示用户的输入， assistant 表示模型的输出）的一种抽象，以便应用于其他大语言模型。SystemMessage对应系统预设，HumanMessage用户输入，AIMessage表示模型输出，使用 ChatMessagePromptTemplate 可以使用任意角色接收聊天消息。

      from langchain.prompts import (
      ChatPromptTemplate,
      PromptTemplate,
      SystemMessagePromptTemplate,
      AIMessagePromptTemplate,
      HumanMessagePromptTemplate,
  )
  from langchain.schema import (
      AIMessage,
      HumanMessage,
      SystemMessage
  )

  def generate_template():
      template="You are a helpful assistant that translates {input_language} to {output_language}."
      system_message_prompt = SystemMessagePromptTemplate.from_template(template)
      human_template="{text}"
      human_message_prompt = HumanMessagePromptTemplate.from_template(human_template)

      chat_prompt = ChatPromptTemplate.from_messages([system_message_prompt, human_message_prompt])
      # [SystemMessage(content='You are a helpful assistant that translates English to Chinese.', additional_kwargs={}), HumanMessage(content='I like Large Language Model', additional_kwargs={}, example=False)]
      final_message = chat_prompt.format_prompt(input_language="English", output_language="Chinese", text="I like Large Language Model").to_messages()
      print(final_message)

  if __name__ == "__main__":
      generate_template()

  • 其他

    • 基于 StringPromptTemplate 自定义提示模板StringPromptTemplate
    • 将Prompt输入与特征存储关联起来(FeaturePromptTemplate)
    • 少样本提示模板（FewShotPromptTemplate）
    • 从示例中动态提取提示词✍️

LLMs

  • LLMs

    将文本字符串作为输入并返回文本字符串的模型（纯文本补全模型），这里重点说下做项目尽量用异步的方式，体验会更好，下面的例子连续10个请求，时间相差接近5s。

      import time
  import asyncio
  from langchain.llms import OpenAI
  def generate_serially():
      llm = OpenAI(temperature=0.9)
      for _ in range(10):
          resp = llm.generate(["Hello, how are you?"])
          print(resp.generations[0][0].text)

  async def async_generate(llm):
      resp = await llm.agenerate(["Hello, how are you?"])
      print(resp.generations[0][0].text)

  async def generate_concurrently():
      llm = OpenAI(temperature=0.9)
      tasks = [async_generate(llm) for _ in range(10)]
      await asyncio.gather(*tasks)

  if __name__ == "__main__":
      s = time.perf_counter()
      asyncio.run(generate_concurrently())
      elapsed = time.perf_counter() - s
      print("\033[1m" + f"Concurrent executed in {elapsed:0.2f} seconds." + "\033[0m")

      s = time.perf_counter()
      generate_serially()
      elapsed = time.perf_counter() - s
      print("\033[1m" + f"Serial executed in {elapsed:0.2f} seconds." + "\033[0m")

  • 缓存

    如果多次请求的返回一样，就可以考虑使用缓存，一方面可以减少对API调用次数节省token消耗，一方面可以加快应用程序的速度。

      from langchain.cache import InMemoryCache
  import time
  import langchain
  from langchain.llms import OpenAI
  llm = OpenAI(model_name="text-davinci-002", n=2, best_of=2)
  langchain.llm_cache = InMemoryCache()
  s = time.perf_counter()
  llm("Tell me a joke")
  elapsed = time.perf_counter() - s
  # executed first in 2.18 seconds.
  print("\033[1m" + f"executed first in {elapsed:0.2f} seconds." + "\033[0m")
  llm("Tell me a joke")
  # executed second in 0.72 seconds.
  elapsed2 = time.perf_counter() - elapsed
  print("\033[1m" + f"executed second in {elapsed2:0.2f} seconds." + "\033[0m")

  • 流式传输

    以打字机效果的方式逐字返回聊天内容

      from langchain.llms import OpenAI
  from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
  llm = OpenAI(streaming=True, callbacks=[StreamingStdOutCallbackHandler()], temperature=0)
  resp = llm("模仿李白的风格写一首唐诗.")
  print(resp)

  • 跟踪 token 消耗情况 流式传输的情况下暂不支持计算，可以考虑内容全部传输完成后用tiktoken库计算

      from langchain.llms import OpenAI
  from langchain.callbacks import get_openai_callback
  llm = OpenAI()
  with get_openai_callback() as cb:
      resp = llm.generate(["模仿李白的风格写一首唐诗."])
      print(resp.generations[0][0].text)
      print(cb)

  • Chat models 将聊天消息列表作为输入并返回聊天消息的模型（对话补全模型）

  • 其他

    • 以json或者yml格式读取保存LLM的（参数）配置（llm.load_llm方法和llm.save方法）
    • 为了节省你的token，还可以在测试过程中使用一个模拟LLM输出的FakeListLLM；还有一个模拟用户输入的HumanInputLLM。
    • 与其他 AI相关基础设施的集成，用到随时查询即可

输出解析器

输出解析器用于构造大语言模型的响应格式，具体通过格式化指令和自定义方法两种方式。

# 格式化指令的方式
from langchain.output_parsers import CommaSeparatedListOutputParser
from langchain.prompts import PromptTemplate, ChatPromptTemplate, HumanMessagePromptTemplate
from langchain.llms import OpenAI
from langchain.chat_models import ChatOpenAI
output_parser = CommaSeparatedListOutputParser()
format_instructions = output_parser.get_format_instructions()
prompt = PromptTemplate(
    template="列出五个 {subject}.\n{format_instructions}",
    input_variables=["subject"],
    partial_variables={"format_instructions": format_instructions}
)
model = OpenAI(temperature=0)
_input = prompt.format(subject="大语言模型的特性")
output = model(_input)
# 可移植性, 可扩展性, 可重用性, 可维护性, 可读性
print(output)
output_parser.parse(output)

虽然内置了 DatetimeOutputParser、EnumOutputParser、PydanticOutputParser等解析器，但是我觉得ResponseSchema的控制自由度更好，但是不易于管理。

from langchain.output_parsers import StructuredOutputParser, ResponseSchema
from langchain.prompts import PromptTemplate, ChatPromptTemplate, HumanMessagePromptTemplate
from langchain.llms import OpenAI
from langchain.chat_models import ChatOpenAI
response_schemas = [
    ResponseSchema(name="answer", description="answer to the user's question"),
    ResponseSchema(name="source", description="source used to answer the user's question, should be a website.")
]
output_parser = StructuredOutputParser.from_response_schemas(response_schemas)
format_instructions = output_parser.get_format_instructions()
prompt = PromptTemplate(
    template="answer the users question as best as possible.\n{format_instructions}\n{question}",
    input_variables=["question"],
    partial_variables={"format_instructions": format_instructions}
)

Data Connection

打通外部数据的管道，包含文档加载，文档转换，文本嵌入，向量存储几个环节。

文档加载

重点包括了csv（CSVLoader），html（UnstructuredHTMLLoader），json（JSONLoader），markdown（UnstructuredMarkdownLoader）以及pdf（因为pdf的格式比较复杂，提供了PyPDFLoader、MathpixPDFLoader、UnstructuredPDFLoader，PyMuPDF等多种形式的加载引擎）几种常用格式的内容解析，但是在实际的项目中，数据来源一般比较多样，格式也比较复杂，重点推荐按需去查看与各种数据源 集成的章节说明，Discord、Notion、Joplin，Word等数据源。

文档拆分

重点关注按照字符递归拆分的方式 RecursiveCharacterTextSplitter ，这种方式会将语义最相关的文本片段放在一起。

文本嵌入

嵌入包含两个方法，一个用于嵌入文档，接受多个文本作为输入；一个用于嵌入查询，接受单个文本。文档中示例使用了OpenAI的嵌入模型text-embedding-ada-002，但提供了很多第三方嵌入模型集成可以按需查看。

from langchain.embeddings import OpenAIEmbeddings
embeddings_model = OpenAIEmbeddings()
# 嵌入文本
embeddings = embedding_model.embed_documents(
    [
        "Hi there!",
        "Oh, hello!",
        "What's your name?",
        "My friends call me World",
        "Hello World!"
    ]
)
len(embeddings), len(embeddings[0])
# 嵌入查询
embedded_query = embedding_model.embed_query("What was the name mentioned in the conversation?")
embedded_query[:5]

向量存储

这个就是对常用矢量数据库（FAISS，Milvus，Pinecone，PGVector等）封装接口的说明，详细的可以前往嵌入专题查看。大概流程都一样：初始化数据库连接信息——>建立索引——>存储矢量——>相似性查询，下面以 Pinecone为例：

from langchain.document_loaders import TextLoader
from langchain.embeddings.openai import OpenAIEmbeddings
import pinecone
loader = TextLoader("../../../state_of_the_union.txt")
documents = loader.load()
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
docs = text_splitter.split_documents(documents)

embeddings = OpenAIEmbeddings()

pinecone.init(
    api_key=PINECONE_API_KEY, 
    environment=PINECONE_ENV,
)
index_name = "langchain-demo"
docsearch = Pinecone.from_documents(docs, embeddings, index_name=index_name)
query = "What did the president say about Ketanji Brown Jackson"
docs = docsearch.similarity_search(query)

数据查询

这节重点关注数据压缩，目的是获得相关性最高的文本带入prompt上下文，这样既可以减少token消耗，也可以保证LLM的输出质量。

from langchain.llms import OpenAI
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain.document_loaders import TextLoader
from langchain.vectorstores import FAISS

documents = TextLoader('../../../state_of_the_union.txt').load()
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
texts = text_splitter.split_documents(documents)
retriever = FAISS.from_documents(texts, OpenAIEmbeddings()).as_retriever()
docs = retriever.get_relevant_documents("What did the president say about Ketanji Brown Jackson")
# 基础检索会返回一个或两个相关的文档和一些不相关的文档，即使是相关的文档也有很多不相关的信息
pretty_print_docs(docs)

llm = OpenAI(temperature=0)
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 about Ketanji Jackson Brown")
pretty_print_docs(compressed_docs)

针对基础检索得到的文档再做一次向量相似性搜索进行过滤，也可以取得不错的效果。

from langchain.retrievers.document_compressors import EmbeddingsFilter

embeddings = OpenAIEmbeddings()
embeddings_filter = EmbeddingsFilter(embeddings=embeddings, similarity_threshold=0.76)
compression_retriever = ContextualCompressionRetriever(base_compressor=embeddings_filter, base_retriever=retriever)

最后一点就是自查询（SelfQueryRetriever）的概念，其实就是结构化查询元数据，因为对文档的元信息查询和文档内容的概要描述部分查询效率肯定是高于全部文档的。

Memory

Chain和Agent是无状态的，只能独立地处理每个传入的查询，Memory 可以管理和操作历史消息。一个带存储的Agent例子如下：

def test_memory():
    search = GoogleSearchAPIWrapper()
    tools = [
        Tool(
            name="Search",
            func=search.run,
            description="useful for when you need to answer questions about current events",
        )
    ]
    prefix = """Have a conversation with a human, answering the following questions as best you can. You have access to the following tools:"""
    suffix = """Begin!"

    {chat_history}
    Question: {input}
    {agent_scratchpad}"""

    prompt = ZeroShotAgent.create_prompt(
        tools,
        prefix=prefix,
        suffix=suffix,
        input_variables=["input", "chat_history", "agent_scratchpad"],
    )
    memory = ConversationBufferMemory(memory_key="chat_history")
    llm_chain = LLMChain(llm=OpenAI(temperature=0), prompt=prompt)
    agent = ZeroShotAgent(llm_chain=llm_chain, tools=tools, verbose=True)
    agent_chain = AgentExecutor.from_agent_and_tools(
        agent=agent, tools=tools, verbose=True, memory=memory
    )
    print(agent_chain.run(input="中国有多少人口?"))

资源推荐

  1. Guidance：微软开源 prompt 编程框架
  2. LangGPT：一种面向大模型的 prompt 编程语言

results matching ""

    No results matching ""