LCEL链
LCEL也被称为LangChain表达式（LangChain Expression Language），是一种用声
明式的方法来链接LangChain组件。
所有可以被链起来的组件，如大语言模型、输出解析器、检索器、提示词模板一般都支持下面三个方法：
stream：流式返回响应的块
invoke：接受输入返回输出
batch：接受批量输入返回输出列表
所以最终组成的链在使用上，也是调用这三个方法。
python
"hljs-keyword">from langchain_community.chat_models "hljs-keyword">import ChatTongyi
"hljs-keyword">from langchain_core.prompts "hljs-keyword">import ChatPromptTemplate
"hljs-keyword">from langchain_openai "hljs-keyword">import ChatOpenAI
"hljs-keyword">from langchain_core.output_parsers "hljs-keyword">import StrOutputParser
"hljs-keyword">import time
"hljs-keyword">from dotenv "hljs-keyword">import load_dotenv
load_dotenv()
# 1. 创建LCEL链
chain = (
ChatPromptTemplate.from_template("用一句话介绍{topic}")
| ChatTongyi()
| StrOutputParser()
)
# 2. 准备批量输入
topics = ["人工智能", "区块链", "量子计算", "基因编辑"]
inputs = [{"topic": topic} "hljs-keyword">for topic in topics] #[{"topic":value},{"topic":value},...]
# 3. 单次调用计时 串行执行，一个个主题执行
start = time.time()
single_results = [chain.invoke({"topic": topic}) "hljs-keyword">for topic in topics]
single_time = time.time() - start
# 4. 批量调用计时
start = time.time()
# 注意事项
# API供应商可能有批量请求限制（如每分钟请求数）
# 输入列表中的所有字典必须有相同的键结构
# 批量处理不适合有状态的操作（如带记忆的对话链）
batch_results = chain.batch(inputs) # 关键批量调用方法
batch_time = time.time() - start
# 5. 结果对比
print(f"\n=== 单次调用耗时: {single_time:.2f}s ===")
"hljs-keyword">for i, res in enumerate(single_results):
print(f"{topics[i]}: {res}")
print(f"\n=== 批量调用耗时: {batch_time:.2f}s (加速 {single_time/batch_time:.1f}x) ===")
"hljs-keyword">for i, res in enumerate(batch_results):
print(f"{topics[i]}: {res}")如何把组件链起来？
最简单最常见的就是用管道操作符“|”
chain = chat_template | client | parser
chain.invoke （｛'language'：'意大利文'，'text'：'朋友啊再见！'｝）
或者使用RunnableSequence
chain = Runnablesequence（ *steps: chat_template, client,parser）
chain.invoke（｛'Language'：'意大利文'，'text'：'朋友啊再见！'｝）
LCEL高级特性与组件
RunnableLambda
很多的时候LangChain组件未提供的功能，需要我们自己写函数实现，但是这个函数我们也需要把它加入链，作为工作流程的一部分，此时就需要使用RunnableLambda
RunnableLambda封装
python
"hljs-keyword">from operator "hljs-keyword">import itemgetter
"hljs-keyword">from langchain_community.chat_models "hljs-keyword">import ChatTongyi
"hljs-keyword">from langchain_core.output_parsers "hljs-keyword">import StrOutputParser
"hljs-keyword">from langchain_core.prompts "hljs-keyword">import ChatPromptTemplate
"hljs-keyword">from langchain_core.runnables "hljs-keyword">import RunnableLambda, chain
"hljs-keyword">from dotenv "hljs-keyword">import load_dotenv
load_dotenv()
# 1.提供自定义函数
# 输入字符串，返回字符串长度
def length_function(text):
"hljs-keyword">return len(text)
#将两个字符串长度的数量相乘
def _multiple_length_function(text1, text2):
"hljs-keyword">return len(text1) * len(text2)
# 输入字典类型，返回字符串长度的乘积
# 方式二：装饰器的方式
@chain
def multiple_length_function(_dict):
"hljs-keyword">return _multiple_length_function(_dict["text1"], _dict["text2"])
prompt = ChatPromptTemplate.from_template("{a} + {b} = ? 计算结果是多少？")
model = ChatTongyi()
out = StrOutputParser()
# 1.通过链来计算字符串的长度
# chain = length_function("hello") 普通函数调用
# chain = RunnableLambda(length_function)
# # 调用执行链
# print(chain.invoke("hello"))
print(RunnableLambda(length_function).invoke("hello"))
# 2.通过大模型计算两个字符串的长度的和
# chain = (
# {"a":itemgetter("k1")| RunnableLambda(length_function) ,
# "b":itemgetter("k2")| RunnableLambda(length_function) }
# |prompt | model | out)
#
# print(chain.invoke({"k1": "hello", "k2": "world"}))
# chain = (
# {"a":itemgetter("k1")| RunnableLambda(length_function) , #5
# # {"text1":"hello","text2":"world"}
# # b=25
# "b":({"text1":itemgetter("k1"),"text2":itemgetter("k2")}|RunnableLambda(multiple_length_function) ) }
# |prompt | model | out)
#
# print(chain.invoke({"k1": "hello", "k2": "world"}))
# 测试装饰器
chain = (
{"a":itemgetter("k1")| RunnableLambda(length_function) , #5
# {"text1":"hello","text2":"world"}
# b=25
"b":({"text1":itemgetter("k1"),"text2":itemgetter("k2")}|multiple_length_function ) }
|prompt | model | out)
print(chain.invoke({"k1": "hello", "k2": "world"}))RunnableParallel/ RunnableMap
把组件链起来，前真我们使用了PunnableSeauence。Sequence的中文意思“顺序，一连串”，也就是所有的组件或多个任务都是按顺序一个个执行的。
但是有时候，我们希望多个任务能够同时执行，这时可以使用RunnableParallel /RunnableMap
python
"hljs-keyword">from langchain_core.runnables "hljs-keyword">import RunnableParallel, RunnableMap, RunnableLambda
def add_one(x: int) -> int:
"hljs-keyword">return x + 1
def mul_two(x: int) -> int:
"hljs-keyword">return x * 2
def mul_three(x: int) -> int:
"hljs-keyword">return x * 3
# 测试parallel并行执行
# chain = RunnableParallel(a=add_one, b=mul_two, c=mul_three)
chain = RunnableMap(a=add_one, b=mul_two, c=mul_three)
# 调用
print(chain.invoke(1))
chain1 = RunnableLambda(add_one)
chain2 = chain1 | RunnableParallel(a=mul_two, b=mul_three)
print(chain2.invoke(2))RunnablePassthrough类允许我们在LangChain的链中传递数据：
确保数据保持不变，直接用于后续步骤的输入。RunnablePassthrough常用在链的第一个位置，用于接收用户的输入（也可以用在中间位置，则用于接收上一步的输出）。
RunnablePassthrough也允许通过assign对数据增强后，再往后传。它会创建一个新的字典，包含原始的所有字段以及你新指定的字段
python
# 原样进行数据传递
"hljs-keyword">from langchain_core.runnables "hljs-keyword">import (
RunnableParallel,
RunnablePassthrough,
RunnableLambda,
)
"hljs-keyword">from setuptools "hljs-keyword">import modified
# chain = RunnableParallel(
# passed=RunnablePassthrough(),
# )
chain = RunnableParallel(
passed=RunnablePassthrough().assign(modified=lambda x: x["k1"] + "!!!")
)
# 调用
print(chain.invoke({"k1": "hello world"})) 版权声明
1 本文章标题： LangChain2
2 本文章地址： https://www.yf2029.com/articles/langchain2
3 本站文章内容仅供学习、交流与界面验证参考，如有版权或引用问题，请联系站点维护者处理。
4 本站不对文中外部链接、第三方资源或读者据此操作产生的结果承担保证责任。
评论
登录后即可发表评论和回复。