LangChain 的链(Chain)
链(Chains)是LangChain的基石,它允许开发者将多个独立的组件连接在一起,形成一个有向无环图(DAG),从而构建出功能更强大、更复杂的LLM驱动型应用。一个链可以是一个简单的顺序执行流程,也可以是包含条件逻辑、并行执行的复杂工作流。
核心概念:Runnable 协议与 LCEL
在LangChain中,所有可被组合的组件(如LLM、ChatModel、PromptTemplate、OutputParser,甚至其他链)都遵循Runnable
协议。这意味着它们都实现了invoke()
、stream()
、batch()
等方法。LangChain Expression Language (LCEL) 是构建和组合这些Runnable组件的核心工具,它提供了一种简洁、强大的方式来定义链。
LCEL 的优点包括:
- 组合性(Composability):所有组件都可组合,使复杂的应用程序易于构建。
- 流式处理(Streaming):支持从链的任何部分进行流式处理,实现更快的响应。
- 异步支持(Async Support):支持异步调用,提高并发性能。
- 并行化(Parallelization):自动并行化 Runnable 序列中的并行步骤,提高效率。
- 日志和可观测性(Logging & Observability):与 LangSmith 深度集成,便于调试和监控。
- 回退机制(Fallback):可以为链中的任何组件配置回退,提高稳健性。
1. 基本链(Basic Chain)
最简单的链通常由“提示模板” -> “语言模型” -> “输出解析器”组成。
工作原理
用户输入首先被 PromptTemplate
格式化,然后发送给 LLM
或 ChatModel
进行处理,最后由 OutputParser
将模型输出转化为我们需要的格式。
# 1. 导入必要的模块
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser# 假设您已设置了OpenAI API Key
# os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"# 2. 定义提示模板
prompt = ChatPromptTemplate.from_template("请用一句话概括以下概念:{concept}")# 3. 定义语言模型
# 默认使用 gpt-3.5-turbo
model = ChatOpenAI(temperature=0.7)# 4. 定义输出解析器
# StrOutputParser 将模型输出直接解析为字符串
output_parser = StrOutputParser()# 5. 组合成链
# 使用 LCEL 的 | 运算符连接组件
chain = prompt | model | output_parser# 6. 调用链
response = chain.invoke({"concept": "大型语言模型"})
print(response)# 示例输出(可能有所不同):
# 大型语言模型是基于大量文本数据训练的AI程序,能够理解、生成和处理人类语言。
2. 顺序链(Sequential Chain)
**顺序链允许你将多个链或可运行对象按顺序连接起来,一个组件的输出作为下一个组件的输入。**这对于需要多步骤处理的任务非常有用。在 LangChain 的新版本中,通常使用 LCEL 的
RunnableSequence
来实现。
工作原理
数据流严格按照定义的顺序从一个组件流向下一个组件。每个组件完成其任务后,将结果传递给序列中的下一个组件。
假设我们要构建一个链,它首先生成一个关于主题的标题,然后根据这个标题生成一个短文本。
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableSequence# 假设您已设置了OpenAI API Keymodel = ChatOpenAI(temperature=0.7)
output_parser = StrOutputParser()# 步骤 1: 生成标题的链
title_prompt = ChatPromptTemplate.from_template("请为主题 '{topic}' 生成一个简洁的标题。")
title_chain = title_prompt | model | output_parser# 步骤 2: 根据标题生成短文本的链
content_prompt = ChatPromptTemplate.from_template("请根据以下标题撰写一个20字左右的短文本:'{title}'")
content_chain = content_prompt | model | output_parser# 组合成顺序链
# RunnableSequence 允许我们将字典作为输入在链中传递,并指定每个步骤的输入/输出
# 这里的 .assign() 方法用于将前一步的输出作为新的键加入到字典中
full_sequential_chain = RunnableSequence({"title": title_chain, # 这一步的输出将被命名为 "title""topic": lambda x: x["topic"] # 保留原始的 topic 输入} | {"output": content_chain.bind(title=lambda x: x["title"]) # 将前一步的 "title" 作为 content_chain 的输入}
)# 或者更简洁地使用 | 运算符和 .assign() 进行链式操作
# 这种方式更符合LCEL的习惯,将上一步的输出直接传递给下一步
full_sequential_chain_lcel = ({"title": title_chain, "topic": lambda x: x["topic"]}| {"output": content_chain.bind(title=lambda x: x["title"])}
)# 调用链
# 注意:这里我们只传入了 'topic',链会负责将 'title_chain' 的输出传递给 'content_chain'
result = full_sequential_chain_lcel.invoke({"topic": "人工智能的未来"})
print(f"生成的标题:{result['title']}")
print(f"生成的短文本:{result['output']}")# 示例输出(可能有所不同):
# 生成的标题:人工智能:塑造未来新格局
# 生成的短文本:人工智能将深刻改变社会面貌,实现智能化生活,开启无限可能的新纪元。
full_sequential_chain_lcel
演示了如何使用 LCEL 的 |
运算符和字典进行链式操作。
- 第一个字典
{"title": title_chain, "topic": lambda x: x["topic"]}
表示并行执行title_chain
来获取title
,同时保留原始的topic
输入。 - 第二个字典
{"output": content_chain.bind(title=lambda x: x["title"])}
表示将上一步生成的title
作为content_chain
的输入,并将其输出命名为output
。
3. 并行链(Parallel Chain)
并行链允许你同时执行多个组件,然后将它们的输出合并。这在需要从不同来源或使用不同方式获取信息,最后再统一处理的场景中非常有用。LCEL 使用
RunnableParallel
或简单的字典语法来实现并行化。
工作原理
多个组件同时接收相同的输入(或输入的不同部分),独立执行,然后它们的输出被收集到一个字典中,供后续处理。
假设我们要同时获取一个主题的中文概括和英文翻译。
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel# 假设您已设置了OpenAI API Keymodel = ChatOpenAI(temperature=0.7)
output_parser = StrOutputParser()# 1. 定义中文概括链
chinese_summary_prompt = ChatPromptTemplate.from_template("请用中文一句话概括:{text}")
chinese_summary_chain = chinese_summary_prompt | model | output_parser# 2. 定义英文翻译链
english_translation_prompt = ChatPromptTemplate.from_template("请将以下文本翻译成英文,只返回翻译结果:{text}")
english_translation_chain = english_translation_prompt | model | output_parser# 3. 组合成并行链
# 使用字典将多个 Runnable 并行化
# 键 'chinese_summary' 和 'english_translation' 将作为最终输出字典的键
parallel_chain = RunnableParallel(chinese_summary=chinese_summary_chain,english_translation=english_translation_chain
)# 或者更简洁的 LCEL 字典语法
parallel_chain_lcel = {"chinese_summary": chinese_summary_chain,"english_translation": english_translation_chain
}# 4. 调用链
text_to_process = "大型语言模型在自然语言处理领域取得了突破性进展。"
results = parallel_chain_lcel.invoke({"text": text_to_process})print(f"中文概括: {results['chinese_summary']}")
print(f"英文翻译: {results['english_translation']}")# 示例输出(可能有所不同):
# 中文概括: 大型语言模型在自然语言处理方面取得了显著的突破。
# 英文翻译: Large language models have made breakthrough progress in the field of natural language processing.
4. 带路由的链(Routing Chain)
路由链允许你根据输入动态地选择执行哪个子链。这在构建复杂的、适应性强的应用时非常有用,例如根据用户问题的类型将其路由到不同的处理逻辑。LCEL 可以通过
RunnableBranch
或自定义函数结合|
运算符来实现路由。
工作原理
一个“路由器”组件接收输入,根据预定义的条件判断或LLM的决策,将输入路由到特定的子链。只有被选中的子链会被执行。
假设我们有一个根据用户提问是关于“事实”还是“创意”来选择不同处理逻辑的链。
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableBranch, RunnableLambda
from langchain_core.runnables.base import RunnablePassthrough# 假设您已设置了OpenAI API Keymodel = ChatOpenAI(temperature=0.7)
output_parser = StrOutputParser()# 1. 事实查询链
fact_prompt = ChatPromptTemplate.from_template("请提供关于'{topic}'的简短事实:")
fact_chain = fact_prompt | model | output_parser# 2. 创意生成链
creative_prompt = ChatPromptTemplate.from_template("请为主题'{topic}'写一个富有创意的短故事(50字以内):")
creative_chain = creative_prompt | model | output_parser# 3. 默认处理链
default_chain = ChatPromptTemplate.from_template("抱歉,我只能处理事实查询或创意生成。您的问题'{topic}'不明确。") | model | output_parser# 4. 路由器:判断输入类型
# 这里我们用一个LLM来判断输入类型,实际上个人感觉 LLM 去判断输入类型会更加灵活,当时整个系统是否可靠非常依赖 LLM 的能力,也可以用正则匹配等方式
router_prompt = ChatPromptTemplate.from_template("""用户提问:'{question}'请判断用户的问题属于以下哪种类型:- FACT (事实查询)- CREATIVE (创意生成)- UNKNOWN (其他或不明确)请只返回类型字符串,例如:FACT
""")
type_classifier_chain = router_prompt | model | StrOutputParser()# 5. 构建路由链
# RunnableBranch 根据条件动态选择分支
# 第一个元组 (条件, 分支)
# 第二个元组 (条件, 分支)
# 最后一个是默认分支
routing_chain = (RunnablePassthrough.assign(question_type=type_classifier_chain.bind(question=RunnablePassthrough()) # 绑定输入,获取类型)| RunnableBranch((lambda x: "FACT" in x["question_type"].upper(), fact_chain.bind(topic=lambda x: x["question"])),(lambda x: "CREATIVE" in x["question_type"].upper(), creative_chain.bind(topic=lambda x: x["question"])),default_chain.bind(topic=lambda x: x["question"]) # 默认分支)
)# 6. 调用链
question1 = "中国的首都是哪里?"
print(f"问题1: {question1}")
print(f"答案1: {routing_chain.invoke({'question': question1})}")question2 = "为'太空旅行'写一个奇幻的短故事。"
print(f"\n问题2: {question2}")
print(f"答案2: {routing_chain.invoke({'question': question2})}")question3 = "你好吗?"
print(f"\n问题3: {question3}")
print(f"答案3: {routing_chain.invoke({'question': question3})}")# 示例输出(可能有所不同):
# 问题1: 中国的首都是哪里?
# 答案1: 中国的首都是北京。
#
# 问题2: 为'太空旅行'写一个奇幻的短故事。
# 答案2: 在遥远的星系,名为埃瑞斯的少女乘坐光舟,穿梭于星云之间。她寻找遗失的星辰之泪,传说它能赋予万物生命,开启宇宙的奇迹。她的旅程充满了未知与神秘。
#
# 问题3: 你好吗?
# 答案3: 抱歉,我只能处理事实查询或创意生成。您的问题'你好吗?'不明确。
这个例子展示了如何使用 RunnableBranch 实现路由。
- 我们首先定义了三个子链:
fact_chain
、creative_chain
和default_chain
。 type_classifier_chain
是一个特殊的链,它的任务是分析用户问题并输出其类型(“FACT”, “CREATIVE”, “UNKNOWN”)。RunnablePassthrough.assign(...)
用于在链的内部步骤中添加新的键值对到输入字典中。这里,我们将type_classifier_chain
的输出作为question_type
添加。RunnableBranch
接收一系列(条件, 分支)
元组和一个可选的默认分支。条件是一个可调用对象(通常是lambda
函数),它接收当前链的输入并返回布尔值。如果条件为True
,则执行对应的分支。
5. 带内存的链(Chain with Memory)
在对话式应用中,模型需要记住之前的对话内容以保持上下文。LangChain 提供了多种内存(Memory)模块来实现这一点,并可以轻松地将其集成到链中。
工作原理
内存模块在链执行前后拦截输入和输出,保存对话历史。当链再次被调用时,内存会将历史记录注入到提示中,以便LLM能够访问完整的上下文。
我们将使用
RunnableWithMessageHistory
来构建一个带内存的聊天机器人。
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory# 假设您已设置了OpenAI API Keymodel = ChatOpenAI(temperature=0.7)
output_parser = StrOutputParser()# 1. 定义带有历史占位符的提示模板
# MessagesPlaceholder("chat_history") 用于注入聊天历史
prompt = ChatPromptTemplate.from_messages([("system", "你是一个友好的AI助手,擅长问答和闲聊。"),MessagesPlaceholder("chat_history"), # 聊天历史的占位符("human", "{question}") # 当前用户问题的占位符
])# 2. 构建核心链(不含内存)
basic_chain = prompt | model | output_parser# 3. 定义会话存储
# 这是一个简单的字典来模拟不同session的聊天历史存储
# 在实际应用中,这会是数据库或Redis等持久化存储
store = {}def get_session_history(session_id: str) -> ChatMessageHistory:if session_id not in store:store[session_id] = ChatMessageHistory()return store[session_id]# 4. 将内存添加到链中
chain_with_history = RunnableWithMessageHistory(basic_chain,get_session_history, # 用于获取特定session历史的函数input_messages_key="question", # 用户输入的键history_messages_key="chat_history" # 历史消息注入到提示中的键
)# 5. 调用链并模拟对话
print("--- Session 1 ---")
config1 = {"configurable": {"session_id": "user123"}}
response1 = chain_with_history.invoke({"question": "你好,我是Alice。"}, config=config1)
print(f"AI: {response1}")response2 = chain_with_history.invoke({"question": "你记得我的名字吗?"}, config=config1)
print(f"AI: {response2}")print("\n--- Session 2 ---")
config2 = {"configurable": {"session_id": "user456"}}
response3 = chain_with_history.invoke({"question": "嗨,我是Bob。"}, config=config2)
print(f"AI: {response3}")response4 = chain_with_history.invoke({"question": "我是谁?"}, config=config2)
print(f"AI: {response4}")# 检查 Alice 的历史,看是否还记得她
print("\n--- 再次与 Alice 对话 ---")
response5 = chain_with_history.invoke({"question": "我叫什么名字?"}, config=config1)
print(f"AI: {response5}")# 示例输出(可能有所不同):
# --- Session 1 ---
# AI: 你好,Alice!很高兴认识你。有什么我可以帮助你的吗?
# AI: 是的,你告诉我你叫Alice。
#
# --- Session 2 ---
# AI: 嗨,Bob!有什么我可以为你服务的吗?
# AI: 你是Bob。
#
# --- 再次与 Alice 对话 ---
# AI: 你是Alice。
RunnableWithMessageHistory 是集成内存的关键。它包装了你的 basic_chain,并在每次调用时:
- 从
get_session_history
获取当前session_id
对应的ChatMessageHistory
。 - 将
ChatMessageHistory
中的对话内容加载到prompt
的chat_history
占位符中。 - 执行
basic_chain
。 - 将新的用户消息和AI响应添加到
ChatMessageHistory
中,以便后续使用。
6. 检索增强生成(RAG)链
RAG 链是 LangChain 最强大的应用模式之一,它结合了信息检索和LLM的生成能力,以克服LLM知识时效性不足和“幻觉”的问题。
工作原理
- 用户查询:接收用户的自然语言问题。
- 检索:使用检索器(通常是向量数据库)根据用户查询在大型文档库中查找最相关的文档片段。
- 增强:将检索到的文档片段作为额外上下文,与用户查询一起输入给LLM。
- 生成:LLM根据提供的上下文生成答案。
下面的库中分别是:openai
(如果使用OpenAI模型)、langchain-openai
、langchain-community
(用于向量存储)、chromadb
(作为向量数据库) 和 tiktoken
(用于token计算)。
pip install -q langchain-openai langchain-community chromadb tiktoken pypdf
from langchain_community.document_loaders import TextLoader # 用于加载文本文件
from langchain_openai import OpenAIEmbeddings, ChatOpenAI # OpenAI 的嵌入模型和聊天模型
from langchain_text_splitters import RecursiveCharacterTextSplitter # 文本分块器
from langchain_community.vectorstores import Chroma # Chroma 向量数据库
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser# 假设您已设置了OpenAI API Key
# os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"# 1. 数据加载与处理
# 为了简化示例,我们创建一个模拟的文档文件
with open("rag_document.txt", "w", encoding="utf-8") as f:f.write("""LangChain 是一个用于开发由大型语言模型驱动的应用程序的框架。它的核心价值在于将复杂的LLM工作流简化为可组合的链。RAG 代表检索增强生成,它通过将信息检索组件与LLM相结合,提高生成答案的准确性和相关性。Chroma 是一个流行的开源向量数据库,常用于存储和检索嵌入向量。构建一个RAG应用通常包括数据加载、文本分割、嵌入、向量存储、检索和生成等步骤。""")loader = TextLoader("rag_document.txt")
docs = loader.load()# 文本分割
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
splits = text_splitter.split_documents(docs)# 2. 嵌入与向量存储
# 使用 OpenAI 的嵌入模型将文本转换为向量
embeddings = OpenAIEmbeddings(model="text-embedding-ada-002")# 将文档块及其嵌入存储到 Chroma 向量数据库中
vectorstore = Chroma.from_documents(documents=splits, embedding=embeddings)# 创建一个检索器
retriever = vectorstore.as_retriever()# 3. 定义提示模板
# 注意这里有一个 {context} 占位符,用于注入检索到的文档内容
rag_prompt = ChatPromptTemplate.from_template("""根据以下上下文信息回答问题:
{context}问题:{question}
""")# 4. 定义语言模型
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)# 5. 组合 RAG 链
rag_chain = ({"context": retriever, "question": RunnablePassthrough()} # 并行获取 context 和 question| rag_prompt # 将 context 和 question 注入到提示中| llm # 调用 LLM| StrOutputParser() # 解析 LLM 输出
)# 6. 调用链
question = "RAG是什么以及它的作用?"
response = rag_chain.invoke(question)
print(response)# 示例输出(可能有所不同):
# RAG代表检索增强生成,它通过将信息检索组件与大型语言模型相结合,提高了生成答案的准确性和相关性。
这个 RAG 链的关键在于 {"context": retriever, "question": RunnablePassthrough()}
。它是一个 RunnableParallel 的隐式表示,做了两件事:
"context": retriever
:调用检索器 (retriever
)。检索器接收rag_chain
的输入(即用户问题),执行检索操作,并返回相关文档。这些文档会被作为context
键的值。"question": RunnablePassthrough()
:将rag_chain
的原始输入(用户问题)直接传递给question
键。
然后,这两个键值对 (context
和 question
) 作为字典传递给 rag_prompt
,提示模板将它们格式化,再传递给 llm
进行生成。
7. 代理(Agents)与工具(Tools)
代理是 LangChain 中最复杂的链类型,它允许 LLM 不仅仅是生成文本,还能进行推理和采取行动。代理可以动态地决定要使用哪些“工具”来完成任务。
工作原理
代理通常运行在一个循环中:
- 输入:接收用户请求。
- 思考(Thought):LLM(代理的“大脑”)根据提示和可用的工具描述进行推理,决定下一步应该做什么:是直接回答,还是需要使用某个工具,使用哪个工具,以及工具的参数是什么。
- 行动(Action):执行LLM选择的工具,并获取工具的输出。
- 观察(Observation):将工具的输出作为观察结果返回给LLM。
- 循环:LLM再次思考,直到它认为任务完成,然后生成最终答案。
from langchain_openai import ChatOpenAI
from langchain import hub
from langchain.agents import AgentExecutor, create_react_agent
# 搜索工具
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.prompts import PromptTemplate# 假设您已设置了OpenAI API Key# 1. 定义工具
# 这里使用 DuckDuckGoSearchRun 作为一个简单的搜索工具
search_tool = DuckDuckGoSearchRun(name="duckduckgo_search")
tools = [search_tool]# 2. 定义提示模板
# create_react_agent 需要一个遵循ReAct模式的提示
# hub.pull("hwchase17/react") 是一个官方维护的ReAct提示模板
prompt = hub.pull("hwchase17/react")# 3. 定义语言模型
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)# 4. 创建代理
agent = create_react_agent(llm=llm, tools=tools, prompt=prompt)# 5. 创建代理执行器
# AgentExecutor 负责运行代理并管理其循环
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)# 6. 调用代理
response = agent_executor.invoke({"input": "奥巴马的生日是哪天?他出生那年发生了什么大事?"})
print(f"\n最终答案: {response['output']}")# 示例输出(verbose=True 会打印思考过程):
# > Entering new AgentExecutor chain...
# I need to find out Barack Obama's birthday and then find major events that happened in that year.
# Action: duckduckgo_search
# Action Input: "Barack Obama birthday"
# Observation: Barack Obama was born on August 4, 1961.
# I have Obama's birthday. Now I need to find major events from 1961.
# Action: duckduckgo_search
# Action Input: "major events 1961"
# Observation: Major events in 1961 include the Bay of Pigs Invasion, the construction of the Berlin Wall, the first human spaceflight by Yuri Gagarin, and the election of John F. Kennedy as U.S. President.
# I have found the major events of 1961. I can now answer the question.
# Final Answer: 巴拉克·奥巴马出生于1961年8月4日。他出生那年发生的重大事件包括猪湾事件、柏林墙的修建、尤里·加加林完成首次载人航天飞行以及约翰·F·肯尼迪当选美国总统。
#
# > Finished chain.
#
# 最终答案: 巴拉克·奥巴马出生于1961年8月4日。他出生那年发生的重大事件包括猪湾事件、柏林墙的修建、尤里·加加林完成首次载人航天飞行以及约翰·F·肯尼迪当选美国总统。
- 工具(Tools):
DuckDuckGoSearchRun
是一个封装了 DuckDuckGo 搜索功能的工具。你可以创建各种自定义工具来访问数据库、API、文件系统等。 create_react_agent
:这是一个方便的函数,用于使用 ReAct (Reasoning and Acting) 范式创建代理。ReAct 代理会交替进行“思考”(LLM的推理)和“行动”(调用工具)。AgentExecutor
:这是运行代理的核心组件。它接收代理的决策,执行工具,将观察结果反馈给代理,并循环直到代理决定给出最终答案。verbose=True
会打印代理的思考过程,这对于调试非常有用。
总结
LangChain 的“链”机制是构建复杂LLM应用的关键。通过 LCEL,你可以像搭积木一样将各种 Runnable
组件(模型、提示、解析器、检索器、工具等)组合起来,形成从简单到复杂的各种工作流。从基本的顺序执行到并行动作,再到带路由的条件分支和能与外部世界交互的智能代理,链提供了无限的可能来构建功能强大、灵活且可维护的LLM应用。