当前位置： 首页 > article >正文

学习路程九 langchain核心组件 Chains

article 2025/3/1 13:56:15

前序

之前完成了Models I/O , 输入提示词，输出解释器。下面学习langchain的另一个核心组件：

链（Chain），也叫链式语言表达式（LCEL），是 LangChain 的核心组件。如果把用 LangChain 构建 AI 大模型应用的过程比作“积木模型”的搭建与拼接，那么 Chain 就是是该模型搭建过程中的骨骼部分，通过它将各模块快速组合在一起就可以快速搭建一个应用。langchain提供的chains使用声明式方法将多个操作链接在一起，形成一个完整的处理流程。常见链的类型有3种：LLMChain，SequentialChain、LLMRouteChain。

官网文档：https://python.langchain.com/docs/concepts/lcel/

大模型链（LLMChain）

LLMChain是最常用的一个

在之前学习的过程种，完成一次调用大致分一下步骤：

#1.创建模版
prompt_template = PromptTemplate.from_template("你是一个产品顾问。请给公司刚生产出来的 {production}，取一个好听的名字和广告语。")
messages = prompt_template.invoke({"production": "充电宝"})

#2.实例化模型
chat_model = ChatDeepSeek(model="deepseek-chat")

#3.通过模版完整的提示词
messages = prompt_template.invoke({"production": "充电宝"})

#4.把提示词传给大模型
response = chat_model.invoke(messages)

# 5.把结果给输出解释器
res= StrOutputParser().invoke(response)

通过 LLMChain(大模型链) 可以直接将数据、Prompt、以及想要应用的大模型串到一起连贯调用。

老版本的定义方式已经被弃用了。

chain = LLMChain（model，prompt）

新的方式如下：

prompt_template = "你是一个产品顾问。请给公司刚生产出来的 {production}，取一个好听的名字和广告语。"

chat_model = ChatDeepSeek(model="deepseek-chat")

chain = PromptTemplate.from_template(prompt_template) | chat_model | StrOutputParser()

chain.invoke({"production": "充电宝"})

这里的chain写法类似于linux的管道符"|"，把上一个的输出作为下一个的输入。

序列链（Sequential Chains）

不同于基本的 LLMChain，Sequential Chain（序列链）是由一系列的链组合而成的，许将多个链按顺序链接在一起，前一个链的输出作为下一个链的输入。序列链有两种类型，一种是单个输入输出，另一个则是多个输入输出。

import os
import langchain
# LangChain相关模块的导入
from langchain.chains import LLMChain
from langchain_deepseek import ChatDeepSeek
from langchain.prompts import ChatPromptTemplate
from langchain.chains.sequential import SequentialChain


os.environ['DEEPSEEK_API_KEY'] = "sk-e243xxx"

# 在全局范围开启详细模式，能将调用大模型时发送的数据打印到控制台，绿色文本
langchain.verbose = True

llm = ChatDeepSeek(model="deepseek-chat", temperature=0.5)

# Chain1 获得内容
prompt1 = ChatPromptTemplate.from_template(
    "请简单介绍一下历史人物: {username}"
)

chain1 = LLMChain(
    llm=llm,
    prompt=prompt1,
    output_key="introduce"
)


# Chain2 根据内容，简明扼要
prompt2 = ChatPromptTemplate.from_template(
    "基于给出的人物信息, 简明扼要，整理出不多于40个字符内容: {introduce}"
)

chain2 = LLMChain(
    llm=llm,
    prompt=prompt2,
    output_key="brief_introduce"
)

# Chain3 把简介翻译成英文
prompt3 = ChatPromptTemplate.from_template(
    "把给出的内容翻译为英文?: {brief_introduce}"
)

chain3 = LLMChain(
    llm=llm,
    prompt=prompt3,
    output_key="english_brief_introduce"
)

# 标准版的序列Chain,SequentialChain,其中每个chain都支持多个输入和输出，
# 根据chains中每个独立chain对象，和chains中的顺序，决定参数的传递，获取最终的输出结果
seq_chain = SequentialChain(
    chains=[chain1, chain2, chain3],
    input_variables=["username"],
    output_variables=["introduce", "brief_introduce", "english_brief_introduce"],
    verbose=True
)

username = "秦始皇"
res = seq_chain(username)
print(res)

在这里插入图片描述

这里还存在一点问题，LLMChain在0.1.17版本就被弃用了，在1.0将会被移除。
所以这里LLMChain 需要更改成官方说的 chain = prompt | llm 这样，但是这样不支持显式的展示输出的key了。怎么展示并显示出运行过程，需要后面研究研究，有结果了再补充回这里。

LLMChain，是有input_keys，但是RunnableSequence源码并没有，所以运行时，prompt | llm 的是一个RunnableSequence 对象，然后又说 RunnableSequence 对象没有input_keys,然后报错，不能单纯只修改成prompt | llm 的样式。
在这里插入图片描述
如果修改成prompt | llm 样式，一下方式也可以运行。但是没有输出过程，只有最终结果

# 方式一，普通序列链调用
seq_chain = chain1 | chain2 | chain3

# 方式二，通过RunnablePassthrough
runnable = RunnablePassthrough.assign(username=lambda inputs: inputs['username']) | chain1 | chain2 | chain3
res = runnable.invoke({'username': '秦始皇'})
print(res)

路由链（RouterChain）

**路由链（RouterChain）**是由LLM根据输入的Prompt去选择具体的某个链。路由链中一般会存在多个Prompt，Prompt结合LLM决定下一步选择哪个链。

路由链一般涉及到2个核心类，LLMRouterChain和MultiPromptChain
LLMRouterChain：使用LLM路由到可能的prompt中

MultiPromptChain ：是一种在自然语言处理（NLP）中使用的技术，主要用于在多个提示词之间进行路由选择。当你有多个提示词并且只想路由到其中一个时，可以使用MultiPromptChain。其基本原理是根据输入的Prompt选择合适的链进行处理‌

使用场景和基本原理
MultiPromptChain通常用于复杂的聊天机器人或NLP应用中，特别是在需要根据用户输入选择不同处理路径的场景。其使用步骤包括：

  1. 准备多个链的Prompt提示词，并封装成链。
  2. 将可能路由到的链封装到destination_chains里。
  3. 构建多提示词和RouterChain，负责选择下一个要调用的链。
  4. 构建默认链，用于未匹配到合适链时的处理‌

以下代码是原本的多路由链使用，但是运行给出警告，MultiPromptChain将会被移除

原使用代码，被废弃

import os
from langchain_deepseek import ChatDeepSeek
import langchain

langchain.verbose = True

# 初始化模型
os.environ['DEEPSEEK_API_KEY'] = "sk-e2xxx"
model = ChatDeepSeek(model="deepseek-chat")

from langchain.chains.router import LLMRouterChain, MultiPromptChain
from langchain.chains.router.llm_router import RouterOutputParser
from langchain.chains.router.multi_prompt_prompt import MULTI_PROMPT_ROUTER_TEMPLATE
from langchain.chains import LLMChain, ConversationChain
from langchain.prompts import PromptTemplate

###### 准备多个提示词封装成链 #####
# 准备3条目的链：一条物理链，一条数学链
# 1. 物理链
physics_template = """
你是一位物理学家，擅长回答物理相关的问题，当你不知道问题的答案时，你就回答不知道。
具体问题如下：
{input}
"""
physics_prompt = PromptTemplate.from_template(physics_template)
# physics_chain = LLMChain(llm=model, prompt=physics_prompt)
physics_chain = physics_prompt | model
# 2. 数学链
math_template = """
你是一个数学家，擅长回答数学相关的问题，当你不知道问题的答案时，你就回答不知道。
具体问题如下：
{input}
"""
math_prompt = PromptTemplate.from_template(math_template)
# math_chain = LLMChain(llm=model, prompt=math_prompt)
math_chain = math_prompt | model

# 3. 英语链
english_template = """
你是一个非常厉害的英语老师，擅长回答英语相关的问题，当你不知道问题的答案时，你就回答不知道。
具体问题如下：
{input}
"""
english_prompt = PromptTemplate.from_template(english_template)
# english_chain = LLMChain(llm=model, prompt=english_prompt)
english_chain = english_prompt | model


###### 将可能路由到的链封装到destination_chains里。 #####
destination_chains = {}
destination_chains["physics"] = physics_chain
destination_chains["math"] = math_chain
destination_chains["english"] = english_chain

######### 构建默认链，用于未匹配到合适链时的处理‌
default_chain = ConversationChain(llm=model, output_key="text")


# 让多路由模板 能找到合适的 提示词模板
destinations_template_str = """
physics:擅长回答物理问题
math:擅长回答数学问题
english:擅长回答英语问题
"""
router_template = MULTI_PROMPT_ROUTER_TEMPLATE.format(
    destinations=destinations_template_str
)

##### 构建多提示词和RouterChain，负责选择下一个要调用的链。 #####
# 通过路由提示词模板，构建路由提示词
router_prompt = PromptTemplate(
    template=router_template,
    input_variables=["input"],
    output_parser=RouterOutputParser(),
)

######### 路由链
router_chain = LLMRouterChain.from_llm(llm=model, prompt=router_prompt)

######### 最终的链
multi_prompt_chain = MultiPromptChain(
    router_chain=router_chain,
    destination_chains=destination_chains,
    default_chain=default_chain,
    verbose=True,
)

# multi_prompt_chain.invoke({"input": "重力加速度是多少？"})
# multi_prompt_chain.invoke("y=x^2+2x+1的导数是多少？")
multi_prompt_chain.run("将以下英文翻译成中文，只输出中文翻译结果：\n The largest community building the future of LLM apps.")

建议使用新的 langgraph,基于图结构的

官网：https://python.langchain.com/docs/versions/migrating_chains/multi_prompt_chain/

from typing import Literal

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableConfig
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict

import os
from langchain_deepseek import ChatDeepSeek
import langchain

langchain.verbose = True

# 初始化模型
os.environ['DEEPSEEK_API_KEY'] = "sk-e243xxx"
llm = ChatDeepSeek(model="deepseek-chat")

# Define the prompts we will route to
prompt_1 = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个数学家，擅长解答数学问题."),
        ("human", "{input}"),
    ]
)
prompt_2 = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个生物学家，擅长解决生物学相关的问题."),
        ("human", "{input}"),
    ]
)

chain_1 = prompt_1 | llm | StrOutputParser()
chain_2 = prompt_2 | llm | StrOutputParser()
default_chain = llm | StrOutputParser()

route_system = "对用户的提问进行合适的分类"
route_prompt = ChatPromptTemplate.from_messages(
    [
        ("system", route_system),
        ("human", "{input}"),
    ]
)


# Define schema for output:
class RouteQuery(TypedDict):
    """Route query to destination expert."""
    destination: Literal["数学", "生物", "其它"]


route_chain = route_prompt | llm.with_structured_output(RouteQuery)


# 定义数据结构
class State(TypedDict):
    query: str
    destination: RouteQuery
    answer: str

query:
async def route_query(state: State, config: RunnableConfig):
    destination = await route_chain.ainvoke(state["query"], config)
    return {"destination": destination}


# 添加节点
async def prompt_1(state: State, config: RunnableConfig):
    return {"answer": await chain_1.ainvoke(state["query"], config)}


async def prompt_2(state: State, config: RunnableConfig):
    return {"answer": await chain_2.ainvoke(state["query"], config)}


async def prompt_3(state: State, config: RunnableConfig):
    return {"answer": await default_chain.ainvoke(state["query"], config)}


# 定义选择节点的逻辑
def select_node(state: State) -> Literal["prompt_1", "prompt_2", "prompt_3"]:
    if state["destination"] == "数学":
        return "prompt_1"
    elif state["destination"] == "生物":
        return "prompt_2"
    else:
        return "prompt_3"


# 定义图的入口和边
graph = StateGraph(State)
graph.add_node("route_query", route_query)
graph.add_node("prompt_1", prompt_1)
graph.add_node("prompt_2", prompt_2)
graph.add_node("prompt_3", prompt_3)

graph.add_edge(START, "route_query")
graph.add_conditional_edges("route_query", select_node)
graph.add_edge("prompt_1", END)
graph.add_edge("prompt_2", END)
graph.add_edge("prompt_3", END)
app = graph.compile()

# 生成图结构
try:
    app.get_graph().draw_mermaid_png(output_file_path="graph.png")
except Exception:
    # This requires some extra dependencies and is optional
    pass


async def run_app():
    result = await app.ainvoke({"query": "人类起源是什么？"})
    print(result["destination"])
    print(result["answer"])


# 使用 asyncio 运行异步函数
import asyncio

asyncio.run(run_app())

在这里插入图片描述

在这里插入图片描述
这块内容感觉比较复杂，只能了解了一下大概实现逻辑，然后好多资料依旧在用MultiPromptChain，LLMChain，就感觉目前学习下来，找到的好多资料都在继续使用即将丢弃的内容,不太新。新的内容学习起来比较困难。

其他

from langchain import chains

点到chains里面，可以看到很多链，可以自行了解，我是看不下去了，推荐的也就是序列链和路由链。
在这里插入图片描述


http://www.kler.cn/a/566015.html

相关文章：

  • 《向华为学习：BEM战略解码》课程大纲
  • [IP] DDR_FIFO(DDR3 用户FIFO接口)
  • PyTorch 常见的损失函数：从基础到大模型的应用
  • 【北京迅为】iTOP-RK3568OpenHarmony系统南向驱动开发-第2章 内核HDF驱动框架架构
  • 咸鱼之王+手游开服搭建架设+内购修复无bug运营版
  • `maturin`是什么：matu rus in python
  • 【含文档+PPT+源码】基于地图系统的校园车辆调度系统的设计与实现
  • Python调用AnythingLLM API使用流输出
  • jenkins使用不同用户管理不同工程
  • 使用通义万相Wan2.1进行视频生成
  • FFmpeg-chapter3-读取视频流（原理篇）
  • 深入探索人工智能的未来：DeepSeek R1与蓝耘智算平台的完美结合
  • 微前端架构深度解码：模块化拆解与联邦宇宙的构建
  • 算法-二叉树篇15-最大二叉树
  • zotero软件的使用和排坑
  • python与C#相比特有的一些语法总结
  • 深入解析 Linux 内核如何解析和获取设备树属性
  • 【汽车ECU电控数据管理篇】HEX文件格式解析篇章
  • iphone上ios设备开启safari开发者debug模式，配合mac电脑使用
  • 华为云设备接入