通过如下设计实现，LCEL 提供了一系列功能，使从基本组件构建复杂链变得容易便捷。

  1. 统一接口：每一个 LCEL对象都实现了 Runnable 接口，该接口定义了一组公共调用方法（invoke, batch, stream, ainvoke, …)。这使得 LCEL 链本身也可以支持这种调用，即每个 LCEL 对象链本身也是一个 LCEL 对象。
  2. 组合原语：LCEL 提供了一系列原语，可以很容易支持组合链，并行组件，添加回退，动态配置链内部运行机制等等。

为了更好理解 LCEL 的价值，看看使用了 LCEL 之前、之后的案例是很有帮助的。下面，我们将选用入门部分的部分基本案例，进行两者之间的对比。

Invoke

在这个最简单的案例中，我们只需传入一个 topic 字符串，并获得一个相关的笑话字符串：

无 LCEL 的实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from typing import List
import openai

prompt_template = "Tell me a short joke about {topic}"
client = openai.OpenAI()

def call_chat_model(messages: List[dict]) -> str:
    response = client.chat.completions.create(
        model="gpt-3.5-turbo", 
        messages=messages,
    )
    return response.choices[0].message.content

def invoke_chain(topic: str) -> str:
    prompt_value = prompt_template.format(topic=topic)
    messages = [{"role": "user", "content": prompt_value}]
    return call_chat_model(messages)

invoke_chain("ice cream")

LCEL实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough


prompt = ChatPromptTemplate.from_template(
    "Tell me a short joke about {topic}"
)
output_parser = StrOutputParser()
model = ChatOpenAI(model="gpt-3.5-turbo")
chain = (
    {"topic": RunnablePassthrough()} 
    | prompt
    | model
    | output_parser
)

chain.invoke("ice cream")

流式处理

如果需要做到流式输出，需要对上面的程序做如下修改：

无 LCEL 实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from typing import Iterator


def stream_chat_model(messages: List[dict]) -> Iterator[str]:
    stream = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=messages,
        stream=True,
    )
    for response in stream:
        content = response.choices[0].delta.content
        if content is not None:
            yield content

def stream_chain(topic: str) -> Iterator[str]:
    prompt_value = prompt.format(topic=topic)
    return stream_chat_model([{"role": "user", "content": prompt_value}])


for chunk in stream_chain("ice cream"):
    print(chunk, end="", flush=True)

LCEL 实现

1
2
for chunk in chain.stream("ice cream"):
    print(chunk, end="", flush=True)

批处理

如果需要批量并行运行一批输入，我们还需要新建一个新的函数：

无 LCEL 实现

1
2
3
4
5
6
7
8
from concurrent.futures import ThreadPoolExecutor

# 线程池并发执行
def batch_chain(topics: list) -> list:
    with ThreadPoolExecutor(max_workers=5) as executor:
        return list(executor.map(invoke_chain, topics))

batch_chain(["ice cream", "spaghetti", "dumplings"])

LCEL 实现

1
chain.batch(["ice cream", "spaghetti", "dumplings"])

异步

异步版本：

无 LCEL 实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
async_client = openai.AsyncOpenAI()

async def acall_chat_model(messages: List[dict]) -> str:
    response = await async_client.chat.completions.create(
        model="gpt-3.5-turbo", 
        messages=messages,
    )
    return response.choices[0].message.content

async def ainvoke_chain(topic: str) -> str:
    prompt_value = prompt_template.format(topic=topic)
    messages = [{"role": "user", "content": prompt_value}]
    return await acall_chat_model(messages)
  
await ainvoke_chain("ice cream")

LCEL 实现

1
chain.ainvoke("ice cream")