文章总结： 本文详细介绍了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_链》
版权声明
本站仅做备份收录，仅供研究与教学参考之用。
读者将信息用于其他用途的，全部法律及连带责任由读者自行承担，本站不承担任何责任。
评论