动手点关注
干货不迷路
本文旨在让无大模型开发背景的工程师或者技术爱好者无痛理解大语言模型应用开发的理论和主流工具，因此会先从与LLM应用开发相关的基础概念谈起，并不刻意追求极致的严谨和完备，而是从直觉和本质入手，结合笔者调研整理及消化理解，帮助大家能够更容易的理解LLM技术全貌，大家可以基于本文衍生展开，结合自己感兴趣的领域深入研究。若有不准确或者错误的地方也希望大家能够留言指正。
本文体系完整，内容丰富，由于内容比较多，分多次连载 。
第一部分 基础概念
1.机器学习场景类别
2.机器学习类型(LLM相关)
3.深度学习的兴起
4.基础模型
第二部分 应用挑战
1.问题定义与基本思路
2.基本流程与相关技术
1）Tokenization与Embbeding
2）向量数据库
3）finetune（微调）
4）模型部署与推理
5）prompt
6）编排与集成
7）其它（预训练等）
第三部分 场景案例
常用参考
2.基本流程与相关技术
点此查看前面内容：
4）编排与集成
经过前面的介绍，我们对大模型应用开发的通常涉及到的重要构成组件都有了介绍，然而这些零散的组件需要有一根”线“将它们串联起来，最终变成用户可以感受到的LLM应用。这里编排集成服务便起到了这样的“线“的作用，它作为粘合剂和应用骨架，在整个应用构成中起到了提纲挈领的作用。
在本章你将学习到：
1）相关理论和应用开发范式
2）Langchain框架介绍
3）LlamaIndex框架介绍
4）Semantic Kernel框架介绍
5） 技术展望
Langchain主要模块介绍
当前LangChain版本是由Langchain-core，Langchain-commuity，Langchain三部分组成，其中Langchain-core的核心是由LCEL构成的，专注于给开发者提供一个灵活、简单的声明式大模型链的构开发协议，而Langchain和Langchain-commuity是构建LLM应用的核心模式及生态实现，其重点是提供标准的，可扩展的接口实现和生态集成的基础模块，包含Models， Retriev al，Agents三类。
LCEL
在langchain项目早期，这一模块是不存在的，但随着应用模式的复杂化，以及具体实现越来越多样，langchain越来越表现的复杂和僵化。为了解决这一问题，以及奠定langchain的标准地位，langchain core的概念逐渐形成，其核心是LCEL（Langchain Expression Language），它在langchain整体架构中定位为协议层。基于它可以有效的构造LLM应用链（Chains），在不用代码变更的情况下，完成从原型开发到生产应用。
官方列举的有关LCEL的一些特点：
1）流式支持：当您使用 LCEL 构建链时，您将获得最佳的 "time-to-first-token"（直到输出第一chunk内容所需的时间）。对于某些链来说，这意味着，例如，我们直接将 LLM 中的token流传输到流式输出解析器，您将以与 LLM 提供者输出原始token相同的速度获得已解析的增量输出chunk。
2）异步支持：使用 LCEL 构建的任何链都可以通过同步 API（例如在 Jupyter 笔记本中进行原型开发时）和异步 API（例如在 LangServe 服务器中）进行调用。这样就能在原型和生产中使用相同的代码，并获得极佳的性能，还能在同一服务器中处理多个并发请求。
3）优化并行执行：只要你的 LCEL 链中有可以并行执行的步骤（例如，从多个检索器中获取文档），我们就会在同步和异步接口中自动执行，以尽可能减少延迟。
4）重试和回退：为 LCEL 链的任何部分配置重试和回退。这是提高链可靠性的好方法。我们目前正在努力为重试/回退添加流支持，这样您就可以在不增加延迟成本的情况下获得更高的可靠性。
5）访问中间结果：对于更复杂的链来说，在生成最终输出之前访问中间步骤的结果往往非常有用。这可以用来让最终用户了解正在发生的事情，甚至只是用来调试链。你可以在每个 LangServe 服务器上对中间结果进行流式处理。
6）输入和输出模式：输入和输出模式为每个 LCEL 链提供了根据链结构推断出的 Pydantic 和 JSONSchema 模式。这可用于验证输入和输出，是 LangServe 不可分割的一部分。
7）无缝的 LangSmith 跟踪集成：随着链变得越来越复杂，了解每一步到底发生了什么变得越来越重要。有了 LCEL，所有步骤都会自动记录到 LangSmith 中，从而实现最大程度的可观察性和可调试性。
8）无缝的 LangServe 部署集成：使用 LCEL 创建的任何链都可以使用 LangServe 轻松部署。
LCEL相较于早期langchain实现方法，最大的优势是更加标准化，原子化，流程化，通过统一的对象接口（都实现Runnable接口）以及对象之间的自由组合，能够更容易应对需求的变化。
使用方法也比较简单，LCEL采用声明式的结构，通过类似linux管道的形式表达chain的逻辑。
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
prompt = ChatPromptTemplate.from_template("tell me a short joke about {topic}")
model = ChatOpenAI()
output_parser = StrOutputParser()
chain = prompt | model | output_parser
chain.invoke({"topic": "ice cream"})
通过LCEL将组件构造成一个链：
chain = prompt | model | output\_parser
“|” 符号类似于 unix 管道操作符，它将不同的组件链在一起，将一个组件的输出作为下一个组件的输入。其pipeline结构如下：
对于相对复杂的RAG来讲，LCEL表达式一样可以完成，如下：
# Requires:
# pip install langchain docarray tiktoken
from langchain.chat_models import ChatOpenAI
from langchain.embeddings import OpenAIEmbeddings
from langchain.prompts import ChatPromptTemplate
from langchain.vectorstores import DocArrayInMemorySearch
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
vectorstore = DocArrayInMemorySearch.from_texts(
["harrison worked at kensho", "bears like to eat honey"],
embedding=OpenAIEmbeddings(),
)
retriever = vectorstore.as_retriever()
template = """Answer the question based only on the following context:
{context}
Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
model = ChatOpenAI()
output_parser = StrOutputParser()
setup_and_retrieval = RunnableParallel(
{"context": retriever, "question": RunnablePassthrough()}
)
chain = setup_and_retrieval | prompt | model | output_parser
chain.invoke("where did harrison work?")
其中，核心的RAG流程同样可以用下面表达式表达：
setup_and_retrieval = RunnableParallel(
{"context": retriever, "question": RunnablePassthrough()}
)
chain = setup_and_retrieval | prompt | model | output_parser
最终形成的Pipeline结构如下：
虽然，LCEL的初衷是为了简化LLM应用构建的复杂度，但它的一些框架概念和约定，比如接口实现和回调钩子会在用户不熟悉的情况下带来一些学习和排错成本。
好处是，Langchain官方 给出了很多常见的LLM应用通过LCEL实现的例子，大家在实现过程中可以对照参考。地址： https://python.langchain.com/docs/expression\_language/cookbook。
基础模块
大模型技术栈和工具百花齐放，但上层的应用模式相对比较统一，为了更好的让开发者避免与底层差异性打交道，因此langchain提供了基础模块的协议实现，第三方可以根据该协议进行适配，这样对于开发者来讲就无须过度关注底层供应商，而只需专注应用本身即可。
- Prompts
在前面介绍里，Prompt的好坏很大程度上决定了大模型输出结果的好坏，但又不是所有开发者都能熟练的掌握Prompt enginering的各种基础，并且，对于应用来讲，它和简单和大模型进行对话也是不同的，为了适配大模型的接口要求以及便于编程，因此langchain针对于LLM应用本身，对围绕Prompt的常见操作进行了一些增强和封装。比如PromptTemplate提供了prompt的模版，可以自动的进行变量的替换，应用开发者无需专门进行处理。
from langchain.prompts import PromptTemplate
prompt_template = PromptTemplate.from_template(
"Tell me a {adjective} joke about {content}."
)
prompt_template.format(adjective="funny", content="chickens")
在针对于COT few-shot场景，为了提高例子选择的有效性，还提供了 Example selectors可以方便的自定义示例的添加。
- Chat Models/LLMs ==================
langchain提供了大量的基座模型的适配，基本上涵盖当下主流的所有模型，可以采用一致的方法添加使用。以OpenAI的LLM为例：
from langchain.llms import OpenAI
llm = OpenAI()
即langchain.llms中选择要导入的模型提供商，初始化使用即可。
除此之外，langchain还提供了请求cache，监控token使用量等常见功能支持，避免开发者从零开始。
- Output parsers ===============
由于大模型的默认输出是非结构化的文本内容。然而，对于程序来讲，更倾向于处理结构化的数据，从而提升应用程序的鲁棒性。而Output Parser便是针对这一需求产生的，它可以将大模型的输出变得结构化。可以说，它是function calling的一个早期兼容实现。官方实现了很多种Parser，常见的有List Parser、Datetime Parser、Enum Parser、Pydantic Parser等等。
下面是一个使用Pydantic Parser解析json对象输出的例子：
from langchain.llms import OpenAI
from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import PromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field, validator
model = OpenAI(model_name="gpt-3.5-turbo-instruct", temperature=0.0)
# Define your desired data structure.
class Joke(BaseModel):
setup: str = Field(description="question to set up a joke")
punchline: str = Field(description="answer to resolve the joke")
# You can add custom validation logic easily with Pydantic.
@validator("setup")
def question_ends_with_question_mark(cls, field):
if field[-1] != "?":
raise ValueError("Badly formed question!")
return field
# Set up a parser + inject instructions into the prompt template.
parser = PydanticOutputParser(pydantic_object=Joke)
prompt = PromptTemplate(
template="Answer the user query.\n{format_instructions}\n{query}\n",
input_variables=["query"],
partial_variables={"format_instructions": parser.get_format_instructions()},
)
# And a query intended to prompt a language model to populate the data structure.
prompt_and_model = prompt | model
output = prompt_and_model.invoke({"query": "Tell me a joke."})
parser.invoke(output)
- Document loaders =================
from langchain.document_loaders import JSONLoader
import json
from pathlib import Path
from pprint import pprint
file_path='./example_data/facebook_chat.json'
data = json.loads(Path(file_path).read_text())
- Document transformers ======================
# This is a long document we can split up.
with open('../../state_of_the_union.txt') as f:
state_of_the_union = f.read()
from langchain.text_splitter import RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter(
# Set a really small chunk size, just to show.
chunk_size = 100,
chunk_overlap = 20,
length_function = len,
add_start_index = True,
)
texts = text_splitter.create_documents([state_of_the_union])
print(texts[0])
print(texts[1])
page_content='Madam Speaker, Madam Vice President, our First Lady and Second Gentleman. Members of Congress and' metadata={'start_index': 0}
page_content='of Congress and the Cabinet. Justices of the Supreme Court. My fellow Americans.' metadata={'start_index': 82}
- Text embedding models ======================
from langchain.embeddings import OpenAIEmbeddings
embeddings_model = OpenAIEmbeddings()
embeddings = embeddings_model.embed_documents(
[
"Hi there!",
"Oh, hello!",
"What's your name?",
"My friends call me World",
"Hello World!"
]
)
len(embeddings), len(embeddings[0])
为了避免重复的进行embedding计算，提高性能降低成本，还提供了CacheBackedEmbeddings的实现。
- Vector stores ==============
作为RAG应用的关键构成，向量数据库对于提升检索召回效果尤为重要。langchain提供了众多向量数据库的集成，开发者可以采用相对统一的方式，结合实际的场景选择不同的实现，比如本地、云等。
import os
import getpass
os.environ['OPENAI_API_KEY'] = getpass.getpass('OpenAI API Key:')
from langchain.document_loaders import TextLoader
from langchain.embeddings.openai import OpenAIEmbeddings
from langchain.text_splitter import CharacterTextSplitter
from langchain.vectorstores import Chroma
# Load the document, split it into chunks, embed each chunk and load it into the vector store.
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())
- Indexing =========
为了方便将文档写入到向量数据库中，langchain提供了标准的建库的API，可以帮助源文档与向量数据库内容保持同步。关键的是，就算前面经过了文档转换，建库的API仍然能够正常工作。这得益于，LangChain建库使用记录管理器（RecordManager）来跟踪写入向量存储的文档。索引内容时，会对每个文档计算哈希值，并在记录管理器中存储以下信息：
- 文档哈希值（页面内容和元数据的哈希值）
- 写入时间
- 源 id - 每份文档都应在元数据中包含信息，以便确定该文档的最终来源
from langchain.embeddings import OpenAIEmbeddings
from langchain.indexes import SQLRecordManager, index
from langchain.schema import Document
from langchain.vectorstores import ElasticsearchStore
Initialize a vector store and set up the embeddings:
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"
)
record_manager.create_schema()
doc1 = Document(page_content="kitty", metadata={"source": "kitty.txt"})
doc2 = Document(page_content="doggy", metadata={"source": "doggy.txt"})
- retriever
retriever，检索器，它作为RAG应用的核心模块，是整个检索过程的载体，是langchain实现的重点，同时，检索器实现了 Runnable 接口，这是 LangChain 表达式语言（LCEL）的基本构件。这意味着它们支持 invoke、ainvoke、stream、astream、batch、abatch、astream_log 调用，其中一个典型应用就是可以更好的观察调试。retriever有大量的实现，比如MultiQueryRetriever、MultiVectorRetriever、self-query retriever等等，它们都是langchain结合生产实际需求不断沉淀的精华所在。
from langchain.chat_models import ChatOpenAI
from langchain.retrievers.multi_query import MultiQueryRetriever
question = "What are the approaches to Task Decomposition?"
llm = ChatOpenAI(temperature=0)
retriever_from_llm = MultiQueryRetriever.from_llm(
retriever=vectordb.as_retriever(), llm=llm
)
Agents应用作为langchain区别于其它框架的一大亮点，是langchain重点关注的领域。Langchain提供了主流的自主agent构建的模式和方法，包括Re-Act、Plan-and-execute 、Self-ask with search、OpenAI assistants等等。
在前面的文章里，我们提到过Agents相较于RAG等LLM Chains应用最大的不同是在Agents中，原本人来规划任务流程和操作变成了由大模型来完成。
有关langchain构建agents的相关细节内容在《 一文探秘LLM应用开发(26)-Prompt(架构模式之Agent框架AutoGPT、AutoGen等) 》，这里就不再赘述。在其官方文档中还有更多的使用案例，这里不再一一介绍。
4）其它
除了前面提到的Model I/O、Retrieval、Agents外，langchain还提供了一些公共的组件支持，比如
Chains、memory和callbacks。
Chains是在LCEL出现之前的应用编排的方法，在更标准、更简单LCEL出现后，官方更推荐使用LCEL，然而，chains中遗留了很多历史实现的一些模式还很有价值，因此开发者也可以直接使用它们。
大模型本身是无记忆的，为了让应用有状态，形成多轮的交互，即支持记忆。langchain提供了Memory模块，它能够帮助应用维持会话信息，自动的将对话历史追加到Prompt中，继而透明的完成大模型记忆的构建，在langchain中还提供了很多优化的手段，如对与历史进行摘要减少历史会话内容过多，导致token消耗或者context window超限的问题。
最后，callbacks是为了提高LLM应用的灵活性，langchain在LLM应用生命周期的各个阶段增设了一些钩子，利用它们用来自定义实现一些逻辑，这对于日志记录、监控、流式传输和其他任务非常有用。
class BaseCallbackHandler:
"""Base callback handler that can be used to handle callbacks from langchain."""
def on_llm_start(
self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any
) -> Any:
"""Run when LLM starts running."""
def on_chat_model_start(
self, serialized: Dict[str, Any], messages: List[List[BaseMessage]], **kwargs: Any
) -> Any:
"""Run when Chat Model starts running."""
def on_llm_new_token(self, token: str, **kwargs: Any) -> Any:
"""Run on new LLM token. Only available when streaming is enabled."""
def on_llm_end(self, response: LLMResult, **kwargs: Any) -> Any:
"""Run when LLM ends running."""
def on_llm_error(
self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any
) -> Any:
"""Run when LLM errors."""
def on_chain_start(
self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any
) -> Any:
"""Run when chain starts running."""
def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> Any:
"""Run when chain ends running."""
def on_chain_error(
self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any
) -> Any:
"""Run when chain errors."""
def on_tool_start(
self, serialized: Dict[str, Any], input_str: str, **kwargs: Any
) -> Any:
"""Run when tool starts running."""
def on_tool_end(self, output: str, **kwargs: Any) -> Any:
"""Run when tool ends running."""
def on_tool_error(
self, error: Union[Exception, KeyboardInterrupt], **kwargs: Any
) -> Any:
"""Run when tool errors."""
def on_text(self, text: str, **kwargs: Any) -> Any:
"""Run on arbitrary text."""
def on_agent_action(self, action: AgentAction, **kwargs: Any) -> Any:
"""Run on agent action."""
def on_agent_finish(self, finish: AgentFinish, **kwargs: Any) -> Any:
"""Run on agent end."""
具体实现也比较简单，开发者只需要覆盖实现相应的方法即可，例如 StdOutCallbackHandler，它 将所有事件记录到stdout ：
class StdOutCallbackHandler(BaseCallbackHandler):
"""Callback Handler that prints to std out."""
def __init__(self, color: Optional[str] = None) -> None:
"""Initialize callback handler."""
self.color = color
def on_chain_start(
self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any
) -> None:
"""Print out that we are entering a chain."""
class_name = serialized.get("name", serialized.get("id", ["<unknown>"])[-1])
print(f"\n\n\033[1m> Entering new {class_name} chain...\033[0m")
def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> None:
"""Print out that we finished a chain."""
print("\n\033[1m> Finished chain.\033[0m")
def on_agent_action(
self, action: AgentAction, color: Optional[str] = None, **kwargs: Any
) -> Any:
"""Run on agent action."""
print_text(action.log, color=color or self.color)
def on_tool_end(
self,
output: str,
color: Optional[str] = None,
observation_prefix: Optional[str] = None,
llm_prefix: Optional[str] = None,
**kwargs: Any,
) -> None:
"""If not the final action, print out observation."""
if observation_prefix is not None:
print_text(f"\n{observation_prefix}")
print_text(output, color=color or self.color)
if llm_prefix is not None:
print_text(f"\n{llm_prefix}")
def on_text(
self,
text: str,
color: Optional[str] = None,
end: str = "",
**kwargs: Any,
) -> None:
"""Run when agent ends."""
print_text(text, color=color or self.color, end=end)
def on_agent_finish(
self, finish: AgentFinish, color: Optional[str] = None, **kwargs: Any
) -> None:
"""Run on agent end."""
print_text(finish.log, color=color or self.color, end="\n")
调用：
from langchain.callbacks import StdOutCallbackHandler
from langchain.chains import LLMChain
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
handler = StdOutCallbackHandler()
llm = OpenAI()
prompt = PromptTemplate.from_template("1 + {number} = ")
# Constructor callback: First, let's explicitly set the StdOutCallbackHandler when initializing our chain
chain = LLMChain(llm=llm, prompt=prompt, callbacks=[handler])
chain.run(number=2)
# Use verbose flag: Then, let's use the `verbose` flag to achieve the same result
chain = LLMChain(llm=llm, prompt=prompt, verbose=True)
chain.run(number=2)
# Request callbacks: Finally, let's use the request `callbacks` to achieve the same result
chain = LLMChain(llm=llm, prompt=prompt)
chain.run(number=2, callbacks=[handler])
输出：
> Entering new LLMChain chain...
Prompt after formatting:
1 + 2 =
> Finished chain.
> Entering new LLMChain chain...
Prompt after formatting:
1 + 2 =
> Finished chain.
> Entering new LLMChain chain...
Prompt after formatting:
1 + 2 =
> Finished chain.
'\n\n3'
小结
Langchain作为当前最热的LLM 应用编排框架，虽然受到了大模型能力增强对其空间的压缩，但它仍然是LLM应用不可或缺的重要支撑。它不断的在扩展着LLM应用的边界，给很多后来的框架和应用开发者以启发，其生态也最为完善，对于想要入门大模型应用开发的朋友来讲，langchain不管未来会不会实际在生产环境使用，但都是值得学习参考的重要案例。
点此查看合集：