【LangChain系列1】【LangChain表达式 (LCEL)】

一、LangChain

1-1、介绍

LangChain是一个框架，用于开发由大型语言模型（LLM）驱动的应用程序。

LangChain 简化了 LLM 应用程序生命周期的每个阶段：

• 开发：使用LangChain的开源构建块和组件构建应用程序。使用第三方集成和模板开始运行。
• 生产化：使用 LangSmith 检查、监控和评估您的链条，以便您可以自信地持续优化和部署。
• 部署：使用 LangServe 将任何链转换为 API。

总结： LangChain是一个用于开发由LLM支持的应用程序的框架，通过提供标准化且丰富的模块抽象，构建LLM的输入输出规范，主要是利用其核心概念chains，可以灵活地链接整个应用开发流程。（即，其中的每个模块抽象，都是源于对大模型的深入理解和实践经验，由许多开发者提供出来的标准化流程和解决方案的抽象，再通过灵活的模块化组合，才得到了langchain）

1-2、LangChain抽象出来的核心模块

想象一下，如果要组织一个AI应用，开发者一般需要？

• 提示词模板的构建，不仅仅只包含用户输入！
• 模型调用与返回，参数设置，返回内容的格式化输出。
• 知识库查询，这里会包含文档加载，切割，以及转化为词嵌入（Embedding）向量。
• 其他第三方工具调用，一般包含天气查询、Google搜索、一些自定义的接口能力调用。
• 记忆获取，每一个对话都有上下文，在开启对话之前总得获取到之前的上下文吧？

由上边的内容，引出LangChain抽象的一些核心模块：

LangChain通过模块化的方式去高级抽象LLM在不同场景下的能力，其中LangChain抽象出的最重要的核心模块如下：‘

• Model I/O ：标准化各个大模型的输入和输出，包含输入模版，模型本身和格式化输出；
• Retrieval ：检索外部数据，然后在执行生成步骤时将其传递到 LLM，包括文档加载、切割、Embedding等；
• Chains ：链条，LangChain框架中最重要的模块，链接多个模块协同构建应用，是实际运作很多功能的高级抽象；
• Memory ： 记忆模块，以各种方式构建历史信息，维护有关实体及其关系的信息；
• Agents ： 目前最热门的Agents开发实践，未来能够真正实现通用人工智能的落地方案；
• Callbacks ：回调系统，允许连接到 LLM 应用程序的各个阶段。用于日志记录、监控、流传输和其他任务；

1-3、特点

LangChain的特点如下：

• 大语言模型(llm): LangChain为自然语言处理提供了不同类型的模型，这些模型可用于处理非结构化文本数据，并且可以基于用户的查询检索信息
• PromptTemplates: 这个特征使开发人员能够使用多个组件为他们的模型构造输入提示。在查询时，开发人员可以使用PromptTemplates为用户查询构造提示模板，之后模板会传递到大模型进行进一步的处理。
• 链:在LangChain中，链是一系列模型，它们被连接在一起以完成一个特定的目标。聊天机器人应用程序的链实例可能涉及使用LLM来理解用户输入，使用内存组件来存储过去的交互，以及使用决策组件来创建相关响应。
• agent: LangChain中的agent与用户输入进行交互，并使用不同的模型进行处理。Agent决定采取何种行动以及以何种顺序来执行行动。例如，CSV Agent可用于从CSV文件加载数据并执行查询，而Pandas Agent可用于从Pandas数据帧加载数据并处理用户查询。可以将代理链接在一起以构建更复杂的应用程序。

1-4、langchain解决的一些行业痛点

在使用大模型的过程中，一些行业痛点：

• 大模型的使用规范以及基于大模型的开发范式不尽相同，当使用一个新模型时，我们往往需要学习新的模型规范。
• 大模型知识更新的滞后性
• 大模型的外部API调用能力
• 大模型输出的不稳定问题，如何稳定输出？
• 大模型与私有化数据的连接方式？

1-5、安装

pip install langchain

二、LangChain表达式——LCEL

2-1、LCEL介绍

LangChain 表达式语言（LCEL）是一种声明式的方法，用于轻松组合多个组件来构建复杂的处理链条。LCEL 设计之初就支持将原型直接投入生产环境，无需更改代码，适用于从简单的“提示+模型”链条到复杂的、包含数百个步骤的链条。

LCEL 的核心特点包括：

• 流式支持：使用 LCEL 构建的链条可以获得最佳的首个令牌时间，即输出的第一块内容出现之前的经过时间。对于某些链条，这意味着可以直接从 LLM 流式传输到流式输出解析器，以与 LLM 提供商输出原始令牌相同的速率获得解析后的增量输出块。
• 异步支持：任何用 LCEL 构建的链条都可以通过同步 API（如在 Jupyter 笔记本中原型设计时）以及异步 API（如在 LangServe 服务器中）调用。
• 优化的并行执行：当 LCEL 链条中有可以并行执行的步骤时（例如，从多个检索器中获取文档），LCEL 会自动执行，以最小化延迟。
• 重试和备选方案：可以为 LCEL 链条的任何部分配置重试和备选方案，以增强链条在大规模生产中的可靠性。

访问中间结果：对于复杂的链条，能够在最终输出产生之前访问中间步骤的结果非常有用，这可以用来通知最终用户或调试链条。

• 输入和输出模式：输入和输出模式为每个 LCEL 链条提供了从链条结构中推断出来的 Pydantic 和 JSONSchema 模式，用于输入和输出的验证。
• 无缝 LangSmith 跟踪集成：所有步骤都会自动记录到 LangSmith，以实现最大的可观察性和可调试性。
• 无缝 LangServe 部署集成：使用 LCEL 创建的任何链条都可以轻松地使用 LangServe 部署。

LCEL 的基本使用示例包括提示（Prompt）+ 模型（Model）+ 输出解析器（OutputParser）。

2-2、基本示例：提示 + 模型 + 输出解析器

基础Demo： 将提示模板和模型链接在一起。使用LCEL将不同的组件组合成一个单一的链条：

from langchain\_core.output\_parsers import StrOutputParser

from langchain\_core.prompts import ChatPromptTemplate

from langchain\_community.chat\_models import ChatZhipuAI



# 定义 Prompt

prompt = ChatPromptTemplate.from\_template("讲一个有关于 {topic} 的笑话")



# 初始化智谱AI的模型实例

# 请替换"your\_api\_key"为你的智谱AI API密钥

# 智谱AI的模型名称根据实际情况填写，例如 "glm-4"

model = ChatZhipuAI(

    model="glm-4",

    temperature=0.5,

    api\_key=""

)



# 定义 Output Parser

output\_parser = StrOutputParser()



# 使用 LCEL 组合链

chain = prompt | model | output\_parser



# 调用链，传递参数

result = chain.invoke({"topic": "冰激凌"})



# 打印结果

print(result)

chain = prompt | model | output_parser

• Prompt: Prompt 是一个 BasePromptTemplate，这意味着它接受一个模板变量的字典并生成一个 PromptValue。PromptValue 是一个包装完成的提示的包装器，可以传递给 LLM（它以字符串作为输入）或 ChatModel（它以消息序列作为输入）。
• Model: 然后将 PromptValue 传递给 model。在这种情况下，我们的 model 是一个 ChatModel，这意味着它将输出一个 BaseMessage。
• OutputParser: 最后，我们将 model 的输出传递给 output_parser，它是一个 BaseOutputParser，意味着它可以接受字符串或 BaseMessage 作为输入。StrOutputParser 简单地将任何输入转换为字符串。(它是一个简单的解析器，从AIMessageChunk中提取content字段)

| 符号类似于 unix 管道操作符，它将不同的组件链接在一起，将一个组件的输出作为下一个组件的输入。在这个链条中，用户输入被传递给提示模板，然后提示模板的输出被传递给模型，然后模型的输出被传递给输出解析器。

2-3、接口

LCEL 通过提供以下功能，使得从基本组件构建复杂链变得容易。它通过提供以下方式实现：

• 统一的接口：每个 LCEL 对象都实现了 Runnable 接口，该接口定义了一组公共的调用方法（invoke、batch、stream、ainvoke，等等）。这使得 LCEL 对象链也自动支持这些调用成为可能。也就是说，每个 LCEL 对象链本身也是一个 LCEL 对象。
• 组合原语：LCEL 提供了一些原语，使得容易组合链，并行化组件，添加回退，动态配置链内部等等。

为了尽可能简化创建自定义链的过程，我们实现了一个 "Runnable" 协议。Runnable 协议已为大多数组件实现。 这是一个标准接口，可以轻松定义自定义链并以标准方式调用它们。 标准接口包括：

• stream: 流式返回响应的块
• invoke: 在输入上调用链
• batch: 在输入列表上调用链

这些方法也有对应的异步方法:

• astream: 异步流式返回响应的块
• ainvoke: 异步在输入上调用链
• abatch: 异步在输入列表上调用链
• astream_log: 异步流式返回中间步骤，以及最终响应
• astream_events: beta 异步流式返回链中发生的事件（在 langchain-core 0.1.14 中引入）

示例：

# 调用链

chain.invoke({"topic": "bears"})



# 异步调用

await chain.ainvoke({"topic": "bears"})



# 批量调用

chain.batch([{"topic": "bears"}, {"topic": "cats"}])



# 异步调用

await chain.abatch([{"topic": "bears"}])



# 流式输出

for s in chain.stream({"topic": "bears"}):

    print(s.content, end="", flush=True)



# 异步的流式输出

async for s in chain.astream({"topic": "bears"}):

    print(s, end="", flush=True)

不同组件的输入类型和输出类型如下：

所有可运行对象都公开输入和输出的模式以检查输入和输出:

• input_schema: 从 Runnable 的结构动态生成的输入 Pydantic 模型
• output_schema: 从 Runnable 的结构动态生成的输出 Pydantic 模型

如下图为2-2获取可运行对象，输入或者是输出产生的描述。（JSONSchema 表示）

附录、ZhiPuAI API

0、安装

pip install zhipu-ai

1、设置API密钥

import os

from langchain\_community.chat\_models import ChatZhipuAI

from langchain\_core.messages import AIMessage, HumanMessage, SystemMessage

from langchain\_core.output\_parsers import StrOutputParser



os.environ["ZHIPUAI\_API\_KEY"] = ""

2、基本用法

chat = ChatZhipuAI(

    model="glm-4",

    temperature=0.5,

)



messages = [

    AIMessage(content="Hi."),

    SystemMessage(content="Your role is a poet."),

    HumanMessage(content="Write a short poem about AI in four lines."),

]



response = chat.invoke(messages)

print(response.content)  # Displays the AI-generated poem

3、流式输出

from langchain\_core.callbacks.manager import CallbackManager

from langchain\_core.callbacks.streaming\_stdout import StreamingStdOutCallbackHandler



streaming\_chat = ChatZhipuAI(

    model="glm-4",

    temperature=0.5,

    streaming=True,

    callback\_manager=CallbackManager([StreamingStdOutCallbackHandler()]),

)

streaming\_chat(messages)

输出如下所示：

4、异步调用

async\_chat = ChatZhipuAI(

    model="glm-4",

    temperature=0.5,

)



response = await async\_chat.agenerate([messages])

print(response)