【LangChain】理论及应用实战（7）：LCEL

一、LCEL简介

LangChain 表达式语言（LangChain Expression Language，简称 LCEL）是 LangChain 框架中的一个核心组件，旨在提供一种简洁、灵活的方式来定义和操作语言模型的工作流。LCEL 允许开发者以声明式的方式构建复杂的语言模型应用，而无需编写大量的样板代码。

以下是 LCEL 的主要优势：

1. 异步、批处理和流支持：采用LCEL构建的任何链都将自动、完全的支持同步、异步、批处理和流等能力。这使得可以在Jupyter中使用同步接口创建链变得很容易，然后将其作为异步流接口进行公开。
2. Fallbacks：由于LLMs的非确定性，使得具备优雅地处理错误的能力变得很重要。通过LCEL，可以轻松地为任何链添加Fallbacks。
3. 并行性：由于LLMs应用涉及（有时长期的）API调用，因此支持并行处理很重要。使用LCEL，任何可以并行处理的组件都会自动并行处理。
4. 无缝集成LangSmith：使用LCEL，所有步骤都会自动记录到LangSmith中，可以最大限度的实现可观察性和可调试性。从而尽量避免因为越来越复杂的链，带来的维护困难。

LCEL包含了多个核心组件，如Prompt, ChatModel, LLM, OutputParser, Retriever, Tool。其为了方便自定义Chain，创造了Runnable协议。Runnable协议适用于大多数组件，是一个标准接口，可以轻松地自定义链并以标准方式调用它们。

我们来看看官网对 Runnable类 的定义：A unit of work that can be invoked, batched, streamed, transformed and composed.
Key Methods（关键方法）：

• invoke/ainvoke: Transforms a single input into an output.
• batch/abatch: Efficiently transforms multiple inputs into outputs.
• stream/astream: Streams output from a single input as it’s produced.
• astream_log: Streams output and selected intermediate results from an input.

LCEL核心组件的运行流程：

二、LCEL示例

LCEL使用“|”运算符链接LangChain应用的各个组件。

2.1 一个简单的示例

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_ollama import OllamaLLMprompt = PromptTemplate.from_template("tell me a short joke about {topic}")
model = OllamaLLM(model="llama3.1:8b")
output_parser = StrOutputParser()chain = prompt | model | output_parser  # LCEL语法result = chain.invoke({"topic": "ice cream"})
print(result)

输出如下：

Why did the ice cream go to therapy?Because it was feeling a little "melted" under pressure!

可以看到，基于LCEL语法形式来定义Chain，比之前我们复写基类的方式来自定义Chain简单了很多。

2.2 RAG Search

主要是实现以下功能：

1. 建立向量数据
2. 使用RAG增强

首先安装依赖包：

pip install --upgrade --quiet langchain langchain-openai faiss-cpu tiktoken

示例代码：

from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_ollama import OllamaLLM, OllamaEmbeddingsvectorstore = FAISS.from_texts(texts=["harrison worked at London"],embedding=OllamaEmbeddings(model="nomic-embed-text:latest")
)
retriever = vectorstore.as_retriever()template = """Answer the question based only on the following context:
{context}Question: {question}
"""prompt = ChatPromptTemplate.from_template(template)model = OllamaLLM(model="llama3.1:8b")chain = ({"context": retriever, "question": RunnablePassthrough()}| prompt| model| StrOutputParser()
)result = chain.invoke("where did harrison work?")
print(result)

输出如下：

London.

【RunnablePassthrough 介绍】，来自官网：https://python.langchain.com/api_reference/core/runnables/langchain_core.runnables.passthrough.RunnablePassthrough.html

• Runnable to passthrough inputs unchanged or with additional keys. This Runnable behaves almost like the identity function, except that it can be configured to add additional keys to the output, if the input is a dict.
• 翻译一下：RunnablePassthrough 可运行以不改变输入或使用其他键传递输入。这个Runnable的行为几乎类似于identity函数，除了如果输入是dict，它可以被配置为向输出添加额外的键。

我们对上面的代码做个简单的解释：

• chain.invoke("where did harrison work?") ：该函数接收到了变量 question = "where did harrison work?"
• "question": RunnablePassthrough()：使用变量 question 为 key，所以它的 value 接收到了 question 的取值，也就是 "where did harrison work?"

我们现在要求其用指定的语言回答我们的问题，可以对上面的代码进行修改，加入 language 相关的代码。如下：

from operator import itemgetter
from langchain_community.vectorstores import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_ollama import OllamaLLM, OllamaEmbeddingsvectorstore = FAISS.from_texts(texts=["harrison worked at London"],embedding=OllamaEmbeddings(model="nomic-embed-text:latest")
)
retriever = vectorstore.as_retriever()template = """Answer the question based only on the following context:
{context}Question: {question}Answer in the following language: {language}
"""prompt = ChatPromptTemplate.from_template(template)model = OllamaLLM(model="llama3.1:8b")chain = ({"context": itemgetter("question") | retriever,"question": itemgetter("question"),"language": itemgetter("language"),}| prompt| model| StrOutputParser()
)result = chain.invoke({"question": "where did harrison work?", "language": "Chinese"})
print(result)

输出如下：

伦敦

三、LCEL下核心组件（Prompt+LLM）的实现

基本构成：PromptTemplate / ChatPromptTemplate -> LLM / ChatModel -> OutputParser

3.1 单链结构

（1）一个简单的样例

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAIprompt = ChatPromptTemplate.from_template("给我讲一个关于{topic}的笑话")
model = ChatOpenAI(temperature=0,model="gpt-3.5-turbo",
)chain = prompt | model
print(chain.invoke({"topic": "狗熊"}))

输出如下：

有一只狗熊想去参加聚会，但他不知道自己应该穿什么衣服。于是，他问了邻居老猫：“喂，我要去参加聚会了，哪种服装比较适合呢？”老猫一脸严肃地说：“我不清楚，但是你最好不要穿那件黄色的衬衣，因为它太显眼了。”

（2）自定义停止输出符

# 自定义停止输出符为\n
chain = prompt | model.bind(stop=["\n"])
print(chain.invoke({"topic": "狗熊"}))

按照上述代码，输出在遇到换行符\n时就会停止输出。

【.bind 函数介绍】，来自官网：https://python.langchain.com/api_reference/core/runnables/langchain_core.runnables.passthrough.RunnablePassthrough.html

• Bind arguments to a Runnable, returning a new Runnable. Useful when a Runnable in a chain requires an argument that is not in the output of the previous Runnable or included in the user input…
• 翻译一下：将参数绑定到Runnable，返回一个新的Runnable。当链中的Runnable需要一个不在前一个Runnable的输出中或包含在用户输入中的参数时，这很有用。

（3）兼容OpenAI函数调用形式

functions = [{"name": "joke","description": "讲笑话","parameters": {"type": "object","properties": {"setup": {"type": "string", "description": "笑话的开头"},"punchline": {"type": "string","description": "爆梗的结尾",},},"required": ["setup", "punchline"],},}
]chain = prompt | model.bind(function_call={"name": "joke"}, functions=functions)
chain.invoke({"topic": "西瓜"})

（4）输出解析器

StrOutputParser ：使用 StrOutputParser 输出解析器后，输出为 str 格式。

from langchain_core.output_parsers import StrOutputParserchain = prompt | model | StrOutputParser()
print(chain.invoke({"topic": "狗熊"}))

【StrOutputParser 介绍】，来自官网：https://python.langchain.com/api_reference/core/output_parsers/langchain_core.output_parsers.string.StrOutputParser.html

• OutputParser that parses LLMResult into the top likely string.
• 翻译一下：OutputParser将LLMResult解析为字符串的形式。

（5）与函数调用混合使用

from langchain.output_parsers.openai_functions import JsonOutputFunctionsParserchain = (prompt| model.bind(function_call={"name": "joke"})| JsonOutputFunctionsParser(key_name="setup")
)print(chain.invoke({"topic": "狗熊"}))

这里，用 JsonOutputParser输出解析器后，输出为 json 格式。其中，key_name 可以指定输出的字段。

JsonOutputParser：Parse the output of an LLM call to a JSON object.

3.2 使用Runnables来连接多链结构

3.2.1 连接多链

from operator import itemgetter  # 获取可迭代对象中指定索引或键对应的元素
from langchain.schema import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_ollama import OllamaLLMprompt1 = ChatPromptTemplate.from_template("{person}来自于哪个城市?")
prompt2 = ChatPromptTemplate.from_template("{city}属于哪个省？用{language}来回答"
)model = OllamaLLM(model="llama3.1:8b")chain1 = prompt1 | model | StrOutputParser()chain2 = ({"city": chain1, "language": itemgetter("language")}  # 获取invoke中的language| prompt2| model| StrOutputParser()
)

我们先来看看 chain1 的输出：

result = chain1.invoke({"person": "马化腾"})
print(result)

马化腾来自深圳。

chain1的输出会被当成chain2的输入，最后得到 chain2 的输出：

result = chain2.invoke({"person": "马化腾", "language": "中文"})
print(result)

马化腾来自于中国广东省深圳市。属于广东省。

3.2.2 多链执行与结果合并

用户的输入会在多个分支分别处理，最后合并结果。

唯物辩证链，给出示例代码：

from operator import itemgetter  # 获取可迭代对象中指定索引或键对应的元素
from langchain.schema import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_ollama import OllamaLLMmodel = OllamaLLM(model="llama3.1:8b")planner = (ChatPromptTemplate.from_template("生成一个关于{input}的论点")| model| StrOutputParser()| {"base_response": RunnablePassthrough()}
)arguments_for = (ChatPromptTemplate.from_template("列出以下内容的优点或积极方面: {base_response}")| model| StrOutputParser()
)arguments_against = (ChatPromptTemplate.from_template("列出以下内容的缺点或消极方面: {base_response}")| model| StrOutputParser()
)final_responder = (ChatPromptTemplate.from_messages([("ai", "{original_response}"),("human", "积极:\n{results_1}\n\n消极:\n{results_2}"),("system", "根据评论生成最终的回复"),])| model| StrOutputParser()
)chain = (planner| {"results_1": arguments_for,"results_2": arguments_against,"original_response": itemgetter("base_response"),}| final_responder
)print(chain.invoke({"input": "生孩子"}))

输出如下：

女性生育孩子是一项艰难但美好的经历。她们承担着创造生命和将其带养大的责任，这需要付出巨大努力和牺牲，但也带来深厚的母爱感激和成就感。女士，作为一个母亲，您一定会有很多不同的体验。但是，无论您是否在育儿中遇到挑战或困难，都应该庆幸自己有了孩子。因为生育孩子不仅给您带来了爱、责任和快乐，也让您有机会培养孩子，并为他们提供安全的家庭环境。当然，育儿也会带来一些不容易改变的负担，如身体疼痛、情绪波动、生活方式的改变等。然而，您也可以从育儿中体验到成就感和责任感，不仅是为您的孩子提供照顾，还有通过育儿来发现自己的能力和潜力。最终，女性生育孩子是一项充满挑战和责任的经历，但也是一个带来深厚母爱感激和成就感的机会。

用户的输入会在多个分支分别处理，最后合并结果。

3.2.3 查询SQL

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParserfrom langchain_community.utilities import SQLDatabasetemplate = """Based on the table schema below, write a SQL query that would answer the user's question:
{schema}Question: 
"""prompt = ChatPromptTemplate.from_template(template)db = SQLDatabase.from_uri("sqlite:///Chinook.db")model = ChatOpenAI(model="gpt-4",temperature=0,
)def get_schema(_):return db.get_table_info()sql_response = (RunnablePassthrough.assign(schema=get_schema)| prompt| model.bind(stop=["\nSQLResult:"])| StrOutputParser()
)sql_response.invoke({"question": "How many artists are there?"})

输出如下：

SELECT COUNT(*) FROM Artist;

可以看到上面示例中输出的是 SQL 查询语句，并没有给出自然语言的结果。下面我们将代码进行一个简单的改造：

template = """Based on the table schema below, write a SQL query that would answer the user's question:
{schema}Question: {question}
SQL Query: {query}
SQL Response: {response}
"""prompt_response = ChatPromptTemplate.from_template(template)
full_chain = (RunnablePassthrough.assign(query=sql_response).assign(schema=get_schema,response=lambda x: db.run(x["query"]),)| prompt_response| model
)
full_chain.invoke({"question": "How many artists are there?"})

输出如下：

There are 275 artists.

3.3 自定义输出解析器

Python编程助手，此处省略。感兴趣的同学可以查看原资料。

四、LCEL添加Memory

4.1 短时记忆

• 基于 ConversationBufferMemory

from operator import itemgetter
from langchain.memory import ConversationBufferMemory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_openai import ChatOpenAImodel = ChatOpenAI()
prompt = ChatPromptTemplate.from_messages([("system", "你是一个乐于助人的机器人"),MessagesPlaceholder(variable_name="history"),("human", "{input}"),]
)memory = ConversationBufferMemory(return_messages=True)chain = (RunnablePassthrough.assign(history=RunnableLambda(memory.load_memory_variables) | itemgetter("history"))| prompt| model
)inputs = {"input": "你好我是Mary"}
response = chain.invoke(inputs)# 保存记忆
memory.save_context(inputs, {'output': response.content})
memory.load_memory_variables({})

4.2 长时记忆

• 基于 Redis

from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_community.chat_models import ChatOpenAI
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory# 基于Redis保存记忆
prompt = ChatPromptTemplate.from_messages([("system", "你是一个擅长{ability}的助手"),MessagesPlaceholder(variable_name="history"),("human", "{question}"),]
)chain = prompt | ChatOpenAI(model="gpt-4-1106-preview", temperature=0)chain_with_history = RunnableWithMessageHistory(chain,# 使用redis存储聊天记录lambda session_id: RedisChatMessageHistory(session_id, url="redis://localhost:6397/0"),input_messages_key="question",history_messages_key="history",
)# 每次调用都会保存聊天记录，需要有对应的session_id
chain_with_history.invoke({"ability": "历史", "question": "中国建都时间最长的城市是哪个?"},config={"configurable": {"session_id": "Mary"}}
)

这里，我们使用了RunnableWithMessageHistory 实现了基于Redis 实现Memory的长时记忆。

五、LCEL：Agent核心组件

5.1 第一个Agent

使用 create_openai_functions_agent 函数直接定义一个简单的 OpenAI 风格的agent，示例代码：

from langchain_openai import ChatOpenAI
from langchain import hub
from langchain.agents import load_tools
from langchain.agents import create_openai_functions_agent  # 不同的agent有不同的创建方式
from langchain.agents import AgentExecutor# 创建LLM
llm = ChatOpenAI(model_name="gpt-4", temperature=0)# 定义agent的prompt
# https://smith.langchain.com/hub/hwchase17/openai-functions-agent
prompt = hub.pull("hwchase17/openai-functions-agent")# 定义工具，加载预制的工具，注意有的工具需要提供LLM
tools = load_tools(["llm-math"], llm=llm)# 创建agent
agent = create_openai_functions_agent(llm, tools, prompt)# 定义agent的执行器，这里注意与老版本的不同
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)agent_executor.invoke({"input": "你好"})

5.2 Agent案例2

实现如下功能：

• 中间步骤处理
• 提示词
• 模型配置（停止符必要的话）
• 输出解析器

from langchain import hub
from langchain.agents import AgentExecutor, tool
from langchain.agents.output_parsers import XMLAgentOutputParser
from langchain_openai import ChatOpenAI# 配置模型
model = ChatOpenAI(model="gpt-4-1106-preview",temperature=0,
)# 使用工具
@tool
def search(query: str) -> str:""" 当需要了解最新的天气的时候，才会使用这个工具。 """return "晴朗, 32摄氏度, 无风"tool_list = [search]# 提示词模板
# https://smith.langchain.com/hub
# Get the prompt to use - you can modify this!
prompt = hub.pull("hwchase17/xml-agent-convo")def convert_intermediate_steps(intermediate_steps):log = ""for action, observation in intermediate_steps:log += (f"<tool>{action.tool}</tool><tool_input>{action.tool_input}"f"></tool_input><observation>{observation}</observation>")return log# 将工具列表插入模板中
def convert_tools(tools):return "\n".join([f"{tool.name}: {tool.description}" for tool in tools])# 定义agent
agent = ({"input": lambda x: x["input"],"agent_scratchpad": lambda x: convert_intermediate_steps(x["intermediate_steps"])},prompt.partial(tools=convert_tools(tool_list)),model.bind(stop=["</tool_input>", "</final_answer>"]),XMLAgentOutputParser()
)# 执行agent
agent_executor = AgentExecutor(agent=agent, tools=tool_list, verbose=True)
result = agent_executor.invoke({"input": "北京今天的天气如何？"})
print(result)