文章总结： 本文详细介绍了LangChain框架中链(Chain)的使用方法，包括组合链和路由链的实现。通过具体代码示例展示了单输入输出链的管道式组合、多输入输出链的变量传递技巧，以及使用RunnableBranch实现条件路由。文档强调了字典传参、延迟处理API限制等实战要点，为开发者提供了构建复杂AI工作流的实用指南。
综合评分： 72
文章分类： AI安全,安全开发,应用安全,安全工具,安全培训

Langchain3_链

原创

羽泪云小栈
羽泪云小栈

羽泪云小栈

2026年4月6日 10:00
陕西

chain-链

https://github.com/ConnectAI-E/LangChain-Tutior/blob/main/python/cn/4.%E6%A8%A1%E5%9E%8B%E9%93%BE.ipynb

【2024吴恩达《基于LangChain的大模型应用开发+构建和评估高级RAG模型应用》带你实现大模型落地！】

https://www.bilibili.com/video/BV1b4421D71Y/?p=4&share_source=copy_web&vd_source=38b4d5bdaa83317738988785494242c4

这次学习组合链、路由链…

之前有条语句是

#老版本用法
chain= LLMChain(llm=myllm, prompt=myprompt)

#新版本
chain=prompt | llm

prompt | llm | jsonparser()  #第三个也就是输出格式的要求

很像linux的管道用法

cat data.txt | grep "error" | sort | uniq -c

都有将左边的输出作为右边的输入的用法

这个数据呢，必须通过字典形式传递变量

比如：

{"role":yly}

组合链-1输入1输出

它可以进行多次chain的结合，当然，需要注意的是有几次chain就发送了几次给llm

prompt_1 = ChatPromptTemplate.from_template(
    "根据我的描述{describe}随机生成一个角色名，必须直接给出名字"
)

prompt_2=ChatPromptTemplate.from_template(
"对这个角色{role}进行一个不超过20字的评价"
)
chain1=prompt_1 | llm
chain2=prompt_2 | llm
chain3=chain1 | chain2
chain3.invoke({"describe":"动漫人物"})

用来在这个步骤中创建角色名和对应描述

这里需要处理的是在chain1和chain2中间，因为对prompt_2指定了{role}变量，需要给它一个。

def rolename(response):
    return {"role":response.content}
chain3=chain1 | rolename | chain2

代码

import os
from dotenv import load_dotenv, find_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
import logging
import time

load_dotenv(find_dotenv())

#logging.basicConfig(level=logging.INFO)
#logging.getLogger("langchain").setLevel(logging.INFO)

llm = ChatOpenAI(
    # This is the default and can be omitted
    api_key=os.environ.get("KEY2"),
    base_url=os.environ.get("base_url2"),
    temperature=0.0,#让预测不让那么随机，偏平稳
    model=os.environ.get("model2")
)

prompt_1 = ChatPromptTemplate.from_template(
    "根据我的描述{describe}随机生成一个角色名，必须直接给出名字"
)

chain_1=prompt_1 | llm

prompt_2=ChatPromptTemplate.from_template(
"对这个角色{role}进行一个不超过20字的评价"
)

def rolename(response):
    #没办法，连发两次会触发429，需要延时
    time.sleep(90)
    print("chain_1的结果是：",response.content)
    return {"role":response.content}

chain_2=prompt_2 | llm
chain_3=chain_1 | rolename | chain_2

result=chain_3.invoke({"describe":input("我:")})
print("结果：\n",result)

我怀疑是现编的一个名字…

组合链(n输入n输出)

多输入多输出怎么弄？

现在我有一段话如下：（来自史铁生《我与地坛》）

但是太阳，他每时每刻都是夕阳也都是旭日。当他熄灭着走下山去收尽苍凉残照之际，正是他在另一面燃烧着爬上山巅布散烈烈朝晖之时。那一天，我也将沉静着走下山去，扶着我的拐杖。有一天，在某一处山洼里，势必会跑上来一个欢蹦的孩子，抱着他的玩具。当然，那不是我。但是，那不是我吗？

1.我想要原文的英文翻译。

2.我想要把英文翻译再用英文总结成一句话作为小总结。

3.原文用的是什么语言

4.用原文的语言翻译这个小总结。

prompt最好格式统一，且要有指向性：

prompt_1 = ChatPromptTemplate.from_template(
    "请将这篇文章段落翻译成英文，只输出一种版本:\n{passage}"
)

prompt_2=ChatPromptTemplate.from_template(
"对这个英文文本用英文总结成一句话:\n{English_passage}"
)

prompt_3=ChatPromptTemplate.from_template(
"请判断以下原文用的是什么语言，只回答(中文、英文等):\n{passage}"
)

prompt_4=ChatPromptTemplate.from_template(
"请将以下总结翻译成{language}语言:\n{summary}"
)

需要用到 RunnablePassthrough.assign，基本上就是输出变量的设置，与变量的传参

因为我的api有频率限制，所以只能延迟触发

def To_delay(response):
    #没办法，连发两次会触发429，需要延时
    time.sleep(90)
    print(f"chain的结果是：",response)
    print()
    return response

代码

import os
from langchain_core.runnables import RunnablePassthrough
from dotenv import load_dotenv, find_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
import logging
import time

load_dotenv(find_dotenv())

#logging.basicConfig(level=logging.INFO)
#logging.getLogger("langchain").setLevel(logging.INFO)

llm = ChatOpenAI(
    # This is the default and can be omitted
    api_key=os.environ.get("KEY2"),
    base_url=os.environ.get("base_url2"),
    temperature=0.0,#让预测不让那么随机，偏平稳
    model=os.environ.get("model2")
)

prompt_1 = ChatPromptTemplate.from_template(
    "请将这篇文章段落翻译成英文，只输出一种版本:\n{passage}"
)

prompt_2=ChatPromptTemplate.from_template(
"对这个英文文本用英文总结成一句话:\n{English_passage}"
)

prompt_3=ChatPromptTemplate.from_template(
"请判断以下原文用的是什么语言，只回答(中文、英文等):\n{passage}"
)

prompt_4=ChatPromptTemplate.from_template(
"请将以下总结翻译成{language}语言:\n{summary}"
)

chain_1=prompt_1 | llm | StrOutputParser()
chain_2=prompt_2 | llm | StrOutputParser()
chain_3=prompt_3 | llm | StrOutputParser()
chain_4=prompt_4 | llm | StrOutputParser()

all_chains=(RunnablePassthrough.assign(
    English_passage=lambda x:To_delay(chain_1.invoke({"passage":x["passage"]}))
)
.assign(
    summary=lambda x:To_delay(chain_2.invoke({
        "English_passage":x["English_passage"]
    }))
)
.assign(
    language=lambda x:To_delay(chain_3.invoke({"passage":x["passage"]}))
)
.assign(
    translated_summary=lambda x:To_delay(chain_4.invoke({
        "language":x["language"],
        "summary":x["summary"]
    }))
)
)

def To_delay(response):
    #没办法，连发两次会触发429，需要延时
    time.sleep(90)
    print(f"chain的结果是：",response)
    print()
    return response

mycontent="但是太阳，他每时每刻都是夕阳也都是旭日。当他熄灭着走下山去收尽苍凉残照之际，正是他在另一面燃烧着爬上山巅布散烈烈朝晖之时。那一天，我也将沉静着走下山去，扶着我的拐杖。有一天，在某一处山洼里，势必会跑上来一个欢蹦的孩子，抱着他的玩具。当然，那不是我。但是，那不是我吗？"
result=all_chains.invoke({"passage":mycontent})
print("passage:\n",result["passage"])
print("English_passage:\n",result["English_passage"])
print("language:\n",result["language"])
print("summary:\n",result["summary"])
print("translated_summary:\n",result["translated_summary"])

注意到每个单独的chain都用到了 StrOutputParser()，结果只会是字符串，其实也就是相当于是message.content

路由链

路由链根据提供的多种不同的prompt模板(比如不同领域、不同场景)，让llm决定选择哪一条链来解决用户的问题。

基本思路就是，让llm去决策用哪个chain，这个叫做router_chain

我们的chain1，chain2，chain3等就组成multi_chain等待被选择触发。

RunnableBranch

但其实方便起见，简单点的场景可以用关键词匹配去触发

这里提供选择分支的是RunnableBranch，多个明确的条件分支，希望一步完成条件判断 + 执行对应分支，用法如下：

from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: isinstance(x, str), lambda x: x.upper()),
    (lambda x: isinstance(x, int), lambda x: x + 1),
    (lambda x: isinstance(x, float), lambda x: x * 2),
    lambda x: "goodbye",
)

branch.invoke("hello") # "HELLO"
branch.invoke(None) # "goodbye"

很像学编程语言时的swich…case语句

它的话，一般适用于关键词匹配，所以它不太需要router_chain

import os
from langchain_core.runnables import RunnablePassthrough
from dotenv import load_dotenv, find_dotenv
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
import time
from typing import Dict, Any

load_dotenv(find_dotenv())

#logging.basicConfig(level=logging.INFO)
#logging.getLogger("langchain").setLevel(logging.INFO)

llm = ChatOpenAI(
    # This is the default and can be omitted
    api_key=os.environ.get("KEY2"),
    base_url=os.environ.get("base_url2"),
    temperature=0.0,#让预测不让那么随机，偏平稳
    model=os.environ.get("model2")
)

prompt_1 = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业的编程助手。请用清晰、带注释的方式回答代码问题。"),
    ("human", "问题：{input}")
])

prompt_2 = ChatPromptTemplate.from_messages([
    ("system", "你是一个翻译专家。请将以下内容翻译成英语（除非用户指定其他语言），只用给一个结果。"),
    ("human", "翻译：{input}")
])

prompt_3 = ChatPromptTemplate.from_messages([
    ("system", "你是一位富有想象力的作家。请根据用户要求创作一段生动、有感情的内容。"),
    ("human", "创作要求：{input}")
])

prompt_4 = ChatPromptTemplate.from_messages([
    ("system", "你是一个数据分析师。请基于给定信息提供逻辑清晰、有依据的分析或建议。"),
    ("human", "分析：{input}")
])

default_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个乐于助人的通用助手。请友好、准确地回答用户问题。"),
    ("human", "{input}")
])

chain_1=prompt_1 | llm | StrOutputParser()
chain_2=prompt_2 | llm | StrOutputParser()
chain_3=prompt_3 | llm | StrOutputParser()
chain_4=prompt_4 | llm | StrOutputParser()
default_chain=default_prompt| llm | StrOutputParser()

branch=RunnableBranch(
     (lambda x: "代码" in x["input"] or "编程" in x["input"] or "def" in x["input"], chain_1),
    (lambda x: "翻译" in x["input"] or "法语" in x["input"] or "英语" in x["input"], chain_2),
    (lambda x: "描写" in x["input"] or "创作" in x["input"] or "故事" in x["input"], chain_3),
    (lambda x: "分析" in x["input"] or "数据" in x["input"] or "统计" in x["input"], chain_4),
    default_chain
)

result = branch.invoke({"input": "翻译这句话:车到山前必有路"})
print(result)

emmm，这很难说是进了chain_2还是default_chain啊，其实想知道的话，可以改一改prompt

prompt_2 = ChatPromptTemplate.from_messages([
    ("system", "你是一个翻译专家。请将以下内容翻译成英语（除非用户指定其他语言），只用给一个结果，并在结果后面输出yly2333。"),
    ("human", "翻译：{input}")
])

RunnableLambda

将一个普通 Python 函数包装成 Runnable，函数返回值会成为下一个 Runnable 的输入。用于实现轻量级、逻辑集中的路由决策（例如返回一个 Runnable 对象或选择 key）

它就可以通过函数，让llm去决策…

...
"""各个目标链"""
destination_chains={
    "代码":chain_1,
    "翻译":chain_2,
    "创作":chain_3,
    "数据":chain_4
}

"""通过这个让llm选择对应的键"""
destinations="""
- 代码:代码编程
- 翻译:文本翻译
- 创作:创作内容
- 数据:数据分析
"""
router_template="""给定一个用户问题，选择最合适的提示模板。
可选提示模板:
{destinations}
判断标准:选择与问题最相关的提示模板。如果都不相关,输出default。
输出格式:只输出提示模板名字，不准输出其它内容
用户问题:{input}
提示模板名:
"""
router_prompt = ChatPromptTemplate.from_template(router_template)

def Pick_router_name(output:str)->str:
    time.sleep(90) #没有调用频率限制就可以去掉
    if output in destination_chains:
        return output
    return "default"

router_chain=router_prompt|llm|StrOutputParser()|Pick_router_name

def Exec_OneRouter(inputs:Dict[str,any])->Dict[str,any]:
    myinput=inputs["input"]
    selected=router_chain.invoke({"destinations":destinations,"input":myinput})
    print(f"路由到了:{selected}")

    if selected in destination_chains:
        result=destination_chains[selected].invoke({"input":myinput})
    else:
        result=default_chain.invoke({"input":myinput})
    return {"input":myinput,"output":result}

final_multi_chains=RunnableLambda(Exec_OneRouter)

result = final_multi_chains.invoke({"input": "描写一个最感人的场景。"})
print(result["output"])
...

比如问一个创作类型的问题：

问到了一个生物问题，归为了默认链：

其实来到这里还是有疑问的，这里一共问了两次llm，1.选哪个chain，2.调用这个chain的结果

那为什么不一次性问完呢？

它的适用场景其实更多用于外接的情况，比如调api、访问数据库等等…

比如，我的订单xxx什么时候发货。

为什么不用一个prompt呢？比如：当用户提到查询订单号时，请执行api如下命令：…” 这不就可以一次性啦？

但是，llm只有文本能力，它不会真的去请求api啊…

那都是先通过llm去判断意图后，按照设定的prompt优化好输出，比如生成”查订单”的命令

再有代码去匹配执行的

然后llm再把这个结果回复给用户(可能美化、总结下结果)，这就是第二次了…

注意

from_template 和 from_message

前者是字符串模板(仅是用户消息)，简单些

后者是对话，结构化

simple_prompt = ChatPromptTemplate.from_template(
    "请帮我总结以下内容：\n{content}\n\n要求：{style}风格"
)

default_prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个乐于助人的通用助手。请友好、准确地回答用户问题。"),
    ("human", "{input}")
])

免责声明：

本文所载程序、技术方法仅面向合法合规的安全研究与教学场景，旨在提升网络安全防护能力，具有明确的技术研究属性。

任何单位或个人未经授权，将本文内容用于攻击、破坏等非法用途的，由此引发的全部法律责任、民事赔偿及连带责任，均由行为人独立承担，本站不承担任何连带责任。

本站内容均为技术交流与知识分享目的发布，若存在版权侵权或其他异议，请通过邮件联系处理，具体联系方式可点击页面上方的联系我。

本文转载自：羽泪云小栈 羽泪云小栈
羽泪云小栈《Langchain3_链》