1. LCEL简介
LCEL(LangChain Expression Language)是一个简单易用的框架，用于构建复杂链条。它提供了统一的接口和组合原语，使得构建链条变得更加容易。每个LCEL对象都实现了Runnable接口，定义了一组常用的调用方法（如invoke、batch、stream、ainvoke等）。因此，LCEL对象的链条也可以自动支持这些调用方法，使得每个LCEL对象的链条本身也是一个LCEL对象。
2. 调用（Invoke）
2.1. 没有使用LCEL
在没有使用LCEL的情况下，可以通过以下代码片段传入一个主题字符串并获取一个笑话字符串。
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")
2.2. 使用LCEL
相比之下，使用LCEL可以更加简洁地实现相同的功能，下面的代码片段展示了如何使用LCEL对象轻松构建链条。
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")
3. 流式处理（Stream）
3.1. 没有使用LCEL
下面的代码片段演示了在不使用LCEL的情况下如何流式处理结果。
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)
stream = stream_chat_model([{"role": "user", "content": prompt_value}])
for chunk in stream:
print(chunk, end="", flush=True)
stream_chain("ice cream")
3.2. 使用LCEL
使用LCEL来流式处理结果更加简便，下面的代码片段展示了如何使用LCEL流式处理结果。
for chunk in chain.stream("ice cream"):
print(chunk, end="", flush=True)
4. 批处理（Batch）
4.1. 没有使用LCEL
下面的代码片段展示了如何在不使用LCEL的情况下并行处理一批输入。
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"])
4.2. 使用LCEL
使用LCEL执行批处理操作非常简便，下面的代码片段展示了如何使用LCEL执行批处理操作。
chain.batch(["ice cream", "spaghetti", "dumplings"])