目标

1
2
3
- 了解LCEL包含的组件，可以调用的方法。

- 链式调用基本形式

LangChain 表达式语言（LCEL）是一种让用户能够更容易地组合不同链条的声明式编程方式。

之前我们介绍了chain 中的组件，这个文档来明确一下组件之间的输入和输出

LCEL 从一开始就是为了支持将原型直接投入生产而设计的，无需修改任何代码。这适用于从最简单的“提示+大语言模型”链条到最复杂的链条（我们已经看到有人成功在生产环境中运行包含数百个步骤的 LCEL 链条）。以下是您可能想使用 LCEL 的几个原因：

  1. 流支持：使用 LCEL 构建链条时，您可以从第一个令牌开始就获得最快的响应时间（即从输出开始到第一个数据块出现的时间）。对于某些链条来说，这意味着我们可以直接从大语言模型向输出解析器发送令牌流，您将以与模型提供者输出原始令牌相同的速度接收到解析后的数据块。

  2. 异步支持：使用 LCEL 构建的任何链条都可以通过同步 API（例如，在 Jupyter 笔记本中测试原型）或异步 API（例如，在 LangServe 服务器中）调用。这使得您可以使用相同的代码进行原型设计和生产，同时还能保持出色的性能，并能在同一服务器上处理多个并发请求。

    同步 API：当您在 Jupyter 笔记本或其他环境中直接调用链条时，您会使用同步 API。这种方式会阻塞当前进程，直到链条执行完毕并返回结果。这适合于原型设计和开发阶段，因为它允许您立即看到每个步骤的结果。

    异步 API：在生产环境中，您通常希望处理多个请求而不会阻塞主进程。这时，您可以使用异步 API。当您通过异步方式调用链条时，请求会立即返回一个唯一标识符或“任务ID”，而不是阻塞并等待结果。您可以在稍后的时间点，使用这个任务ID来检查链条的状态或获取结果。这种方式允许服务器同时处理多个请求，从而提高了效率和性能。

  3. 并行执行优化：当 LCEL 链条中有可以并行执行的步骤（例如，从多个检索器获取文档）时，系统会自动进行优化，无论是在同步还是异步接口中，都能实现最小的延迟。

  4. 重试和回退：您可以为 LCEL 链条的任何部分配置重试和回退策略。这是在大规模应用中提高链条可靠性的好方法。我们目前正在为重试和回退添加流支持，这样您就可以在不增加延迟的情况下获得更高的可靠性。

  5. 访问中间结果：对于复杂的链条，通常在最终输出完成之前访问中间步骤的结果是非常有用的。这可以用来通知最终用户正在发生的事情，或者只是用来调试您的链条。您可以流式传输中间结果，并且可以在每个 LangServe 服务器上使用。

  6. 输入和输出架构：LCEL 链条都有 Pydantic 和 JSONSchema 架构，这些架构是从链条的结构中自动推断出来的。这可以用来验证输入和输出，并且是 LangServe 不可或缺的一部分。

  7. 与 LangSmith 的无缝集成：随着链条变得越来越复杂，了解每个步骤的具体情况变得越来越重要。使用 LCEL，所有步骤都会自动记录到 LangSmith 中，以便进行最大程度的监控和调试。

  8. 与 LangServe 的无缝部署集成：使用 LCEL 创建的任何链条都可以轻松地通过 LangServe 部署。

快速开始

基础链: prompt + model + output parser

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# 环境设置
os.environ["QIANFAN_ACCESS_KEY"] = os.getenv('MY_QIANFAN_ACCESS_KEY')
os.environ["QIANFAN_SECRET_KEY"] = os.getenv('MY_QIANFAN_SECRET_KEY')

# chatModel
model = QianfanChatEndpoint(
    model="ERNIE-Bot-4"
)

# 提示词
prompt = ChatPromptTemplate.from_template("给我讲一个关于{topic}的笑话")

# 输出解析器
output_parser = StrOutputParser()

# 基本链
chain = prompt | model | output_parser

res = chain.invoke({"topic": "冰淇淋"})
print(res)
----
# 输出
当然可以，这是一个关于冰淇淋的笑话：...
----

这里的 “|” 符号有点像 Unix 系统中的管道操作符，它把不同的部分连接起来，把一个部分的输出作为下一个部分的输入。在这个流程中，用户输入首先传给提示模板，然后提示模板的输出传给模型，模型的输出再传给输出解析器。我们来逐一看看每个部分，以便更好地理解整个流程。

  1. 提示词（Prompt） 提示是一个基础的提示模板（BasePromptTemplate），这意味着它接收一个模板变量的字典，并生成一个提示值（PromptValue）。提示值是对已完成提示的包装，可以传递给大语言模型（LLM，它接受字符串作为输入）或聊天模型（ChatModel，它接受一系列的消息作为输入）。它能够与任何语言模型类型配合工作，因为它定义了生成基础消息（BaseMessages）和生成字符串的逻辑。
1
2
3
4
5
6
7
8
9
10
11
12
# 提示词
prompt_value = prompt.invoke({"topic": "冰淇淋"})

cp_value = ChatPromptValue(messages=[HumanMessage(content='给我讲一个关于冰淇淋的笑话')])

# prompt_value 和 cp_value 相同

message = prompt_value.to_messages()
# 输出 [content='给我讲一个关于冰淇淋的笑话']

pr0_str = prompt_value.to_string()
# 输出  'Human: 给我讲一个关于冰淇淋的笑话'
  1. 模型（Model） 提示词接着被传递给模型。在这个例子中，我们的模型是一个聊天模型（ChatModel），这意味着它将输出一个基础消息（BaseMessage）。
1
2
3
4
5
6
model = QianfanChatEndpoint(
    model="ERNIE-Bot-4"
)
...
message = model.invoke(prompt_value)
# message 是一个 AIMessage类的对象
  1. 输出解析器（Output parser） 最后，我们将模型输出传递给输出解析器，它是一个基础的输出解析器（BaseOutputParser），这意味着它接受字符串或基础消息（BaseMessage）作为输入。StrOutputParser 特别是将任何输入简单转换为字符串。
1
2
3
4
5
6
# 解析器
output_parser = StrOutputParser()
res = output_parser.invoke(message)
print(res)

# 输出 ：当然可以，这是一个关于冰淇淋的笑话...
  1. 整个流程（Entire Pipeline） 让我们一步一步地跟随这个过程：
  • 首先，我们输入用户对所需主题的输入，例如 **{“topic”: “ice cream”}**。
  • 提示组件接收用户输入，并使用该主题来构建提示，然后生成一个 PromptValue。
  • 模型组件接收生成的提示，并将其传递给 OpenAI 大语言模型进行评估。模型生成的输出是一个 ChatMessage 对象。
  • 最后，输出解析器组件接收一个 ChatMessage，并将其转换成一个 Python 字符串，这个字符串从 invoke 方法中返回。

可以这么调用么？

1
2
3
4
# res = chain.invoke({"topic": "冰淇淋"})

# 注意，由于需要指定参数，下面这样调用是不行的
res = chain.invoke("冰淇淋")  <--------- 不行

需要指定参数传给谁 （RunnablePassthrough() 从输入接收参数，传递给topic）

1
2
3
4
5
6
7
chain = (
        {"topic": RunnablePassthrough()}
        | prompt
        | model
        | output_parser
)
res = chain.invoke("冰淇淋") <----------可以

检索流程 （了解一下，后面详解）

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
# 千帆嵌入模型
embeddings_model = QianfanEmbeddingsEndpoint(model="bge_large_en", endpoint="bge_large_en")

# 载入数据库
vector_store = Chroma(persist_directory="D:\\LLM\\my_projects\\chroma_db", embedding_function=embeddings_model)

# 创建检索器
retriever = vector_store.as_retriever()

# 提示词
template = """Answer the question based only on the following context:
{context}

Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)

chain = (
        {"context": retriever, "question": RunnablePassthrough()}
        | prompt
        | model
        | output_parser
)
res = chain.invoke("how can langsmith help with testing?")
print(res)

完整代码

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
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
import os

from langchain_community.chat_models.baidu_qianfan_endpoint import QianfanChatEndpoint
from langchain_community.embeddings import QianfanEmbeddingsEndpoint
from langchain_community.vectorstores.chroma import Chroma
from langchain_core.messages import HumanMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompt_values import ChatPromptValue
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

if __name__ == '__main__':

    os.environ["QIANFAN_ACCESS_KEY"] = os.getenv('MY_QIANFAN_ACCESS_KEY')
    os.environ["QIANFAN_SECRET_KEY"] = os.getenv('MY_QIANFAN_SECRET_KEY')

    model = QianfanChatEndpoint(
        model="ERNIE-Bot-4"
    )

    prompt = ChatPromptTemplate.from_template("给我讲一个关于{topic}的笑话")

    output_parser = StrOutputParser()

    chain = prompt | model | output_parser

    # res = chain.invoke({"topic": "冰淇淋"})

    # 注意，由于需要指定参数，下面这样调用是不行的
    # res = chain.invoke("冰淇淋")

    chain = (
            {"topic": RunnablePassthrough()}
            | prompt
            | model
            | output_parser
    )
    # res = chain.invoke("冰淇淋")

    # print(res)
    pass

    # 提示词
    prompt_value = prompt.invoke({"topic": "冰淇淋"})
    prompt_value

    cp_value = ChatPromptValue(messages=[HumanMessage(content='给我讲一个关于冰淇淋的笑话')])

    message = prompt_value.to_messages()

    pr0_str = prompt_value.to_string()
    pass

    # 模型
    message = model.invoke(prompt_value)
    print(message)
    pass

    # 解析器
    output_parser = StrOutputParser()
    res = output_parser.invoke(message)
    print(res)
    pass

    # 检索chain

    # 千帆嵌入模型
    embeddings_model = QianfanEmbeddingsEndpoint(model="bge_large_en", endpoint="bge_large_en")
    # 载入本地向量数据库 Chr
    vector_store = Chroma(persist_directory="D:\\LLM\\my_projects\\chroma_db", embedding_function=embeddings_model)

    # 创建检索器
    retriever = vector_store.as_retriever()

    # 提示词
    template = """Answer the question based only on the following context:
    {context}

    Question: {question}
    """
    prompt = ChatPromptTemplate.from_template(template)

    chain = (
            {"context": retriever, "question": RunnablePassthrough()}
            | prompt
            | model
            | output_parser
    )
    res = chain.invoke("how can langsmith help with testing?")
    print(res)