一、前言
尽管现在的大语言模型已经非常强大，可以解决许多问题，但在处理复杂情况时，仍然需要进行多个步骤或整合不同的流程才能达到最终的目标。然而，现在可以利用langchain来使得模型的应用变得更加直接和简单。
LCEL是什么？
LCEL是一种非常灵活和强大的语言,可以帮助您更方便地构建复杂的应用程序。它提供了很多有用的功能,使得您可以更快速地完成开发任务。
上一篇开源模型应用落地-LangChain高阶-LCEL-表达式语言（六）我们通过两个案例学习了运行自定义函数和接收可运行配置，本篇开始继续学习LCEL的高级用法：传递传入，即实现在不改变或添加额外键的情况下改变键值。
二、术语
2.1.LCEL
是一种用于构建复杂链式组件的语言。它提供了一些非常实用的功能,比如流式处理、并行化和日志记录。它的使用方式很简单:只需要使用 "|" 符号将不同的组件连接起来,就可以形成一个链式结构。这个 "|" 符号就像是Unix中的管道操作符,可以将一个组件的输出转换成下一个组件的输入,从而实现数据的传递和处理。
2.2. Passthrough: Pass through inputs（传递输入）
在不改变或添加额外键的情况下传递输入。它通常与RunnableParallel结合使用，用于分配数值到字典的新键。
三、前提条件
3.1.安装虚拟环境
conda create -n langchain python=3.10
conda activate langchain
pip install --quiet langchain-core langchain-community langchain-openai -i https://pypi.tuna.tsinghua.edu.cn/simplePS：--quiet: 这个选项会让输出信息更简洁,只显示必要的信息,不会输出太多冗余信息。
四、技术实现
4.1.基础示例
不更改传递输入，将数据传递到映射中的新键。
# -*- coding: utf-8 -*-
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
def get_5A_Scenic_Spots(prompt):
print(prompt['prompt'])
return prompt['prompt']+"特色美食有哪些？"
if __name__ == '__main__':
runnable = RunnableParallel(
passed=RunnablePassthrough(),
modified=get_5A_Scenic_Spots,
)
result = runnable.invoke({"prompt": "广州特色景点有哪些?"})
print(result)调用结果：
说明：
原输入的prompt为：广州特色景点有哪些？修改后的prompt为：广州的特色景点有哪些？特色美食有哪些？
4.2.综合示例
# -*- coding: utf-8 -*-
import os
from operator import itemgetter
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
from langchain_openai import ChatOpenAI
os.environ["OPENAI_API_KEY"] = 'sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx' # 你的Open AI Key
def get_rainfall(prompt):
print('修改前的prompt：'+prompt['question'])
prompt_new = prompt['question'].replace("天气如何？", "降雨量如何？")
print('修改后的prompt：' + prompt_new)
return
if __name__ == '__main__':
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0, max_tokens=512)
context = "6月9日，广州市出现中雨局部大雨，全市平均雨量11.2毫米，增城区仙村镇录得最大累计雨量35.6毫米和最大小时雨强27.1毫米。各区最低气温23~24℃，最高气温25~28℃。10日-12日副高加强，我市降雨有所减弱，以午后雷阵雨为主，间中雨势较大。高考期间赴考时段可能与雷雨天气叠加，交通拥堵气象风险高，建议家长和考生提前规划好赶考路线，预留充足时间。午后体感略闷热，建议考场做好通风降温工作，医疗等相关部门做好保障工作。"
question = "6月9日广州市的天气如何？"
template = """Answer the question based only on the following context:
{context}
Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)
chain = (
{"context": itemgetter("context"), "question": RunnablePassthrough()}
| prompt
| model
| StrOutputParser()
)
runnable = RunnableParallel(
chain=chain,
modified=get_rainfall,
)
# resp = chain.invoke({"context": context, "question": question})
# print(resp)
resp = runnable.invoke({"context": context, "question": question})
print(resp['chain'])
调用结果：
说明：
在RunnableParallel中将询问天气的Prompt更改为询问降雨量，使得推理结果为：6月9日广州市出现中雨局部大雨，全市平均雨量11.2毫米，增城区仙村镇录得最大累计雨量35.6毫米和最大小时雨强27.1毫米。各区最低气温23~24℃，最高气温25~28℃。
五、附带说明
5.1.RunnableParallel组件
是一个特殊的组件，它允许并行运行多个处理步骤。当想要同时执行多个任务，而这些任务之间不需要相互等待或依赖对方的结果时，使用RunnableParallel就非常有用，这可以显著提高处理速度，特别是在处理需要大量计算资源的任务时。
RunnableParallel可以轻松地并行执行多个Runnables，并将这些Runnables的输出作为映射返回。
5.2.operator.itemgetter 作用
是 Python 标准库中的一个函数,它可以用于从序列类型(如列表、元组等)中提取特定项目。它的主要用途如下:
从列表中提取特定项目:
from operator import itemgetter
data = [('Alice', 25, 'Engineer'), ('Bob', 30, 'Manager'), ('Charlie', 35, 'Director')]
names = itemgetter(0)
ages = itemgetter(1)
jobs = itemgetter(2)
print(list(map(names, data))) # Output: ['Alice', 'Bob', 'Charlie']
print(list(map(ages, data))) # Output: [25, 30, 35]
print(list(map(jobs, data))) # Output: ['Engineer', 'Manager', 'Director']根据特定项目对列表进行排序:
from operator import itemgetter
data = [('Alice', 25, 'Engineer'), ('Bob', 30, 'Manager'), ('Charlie', 35, 'Director')]
sorted_by_age = sorted(data, key=itemgetter(1))
print(sorted_by_age)
# Output: [('Alice', 25, 'Engineer'), ('Bob', 30, 'Manager'), ('Charlie', 35, 'Director')]
sorted_by_name = sorted(data, key=itemgetter(0))
print(sorted_by_name)
# Output: [('Alice', 25, 'Engineer'), ('Bob', 30, 'Manager'), ('Charlie', 35, 'Director')]在字典中提取特定项目:
from operator import itemgetter
person = {'name': 'Alice', 'age': 25, 'job': 'Engineer'}
name_getter = itemgetter('name')
age_getter = itemgetter('age')
job_getter = itemgetter('job')
print(name_getter(person)) # Output: 'Alice'
print(age_getter(person)) # Output: 25
print(job_getter(person)) # Output: 'Engineer'在元组中提取特定项目:
from operator import itemgetter
point = (10, 20, 30)
x_getter = itemgetter(0)
y_getter = itemgetter(1)
z_getter = itemgetter(2)
print(x_getter(point)) # Output: 10
print(y_getter(point)) # Output: 20
print(z_getter(point)) # Output: 30
