Langchain模块Chain组件-多种chain

 2025/12/18 

路由chain

路由链是LangChain中用于智能决策和分发的核心机制，它可以根据输入内容自动选择最合适的处理路径。

from langchain_core.runnables import RunnablePassthrough, RunnableLambda, RunnableBranch
from langchain.prompts import ChatPromptTemplate,SystemMessagePromptTemplate,HumanMessagePromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_deepseek import ChatDeepSeek
from dotenv import load_dotenv

load_dotenv()

# from langchain.globals import set_debug
# set_debug(True)

# 创建领域链
physics_chain = (
    ChatPromptTemplate.from_template("你是物理专家。问题：{input}")
    | ChatDeepSeek(model="deepseek-chat")
    | StrOutputParser()
)

math_chain = (
    ChatPromptTemplate.from_template("你是数学专家。问题：{input}")
    | ChatDeepSeek(model="deepseek-chat")
    | StrOutputParser()
)

general_chain = (
    ChatPromptTemplate.from_template("你是通用助手。问题：{input}")
    | ChatDeepSeek(model="deepseek-chat")
    | StrOutputParser()
)

# 分类函数,进行问题的分类
def classify_input(input_text: str) -> str:
    prompt = ChatPromptTemplate.from_messages([
        SystemMessagePromptTemplate.from_template("你是一个问题分类专家"),
        HumanMessagePromptTemplate.from_template("{input}。请分类这个问题：如果是物理问题返回字符串physics，数学问题返回字符串math，否则返回字符串general。"),
    ])
    chain = prompt | ChatDeepSeek(model="deepseek-chat") | StrOutputParser()
    return chain.invoke({"input": input_text}).strip().lower()

#  后处理函数（可选
def post_processor(response: str) -> str:
    """添加结果前缀"""
    return f"最终回答：{response}"

# 路由链
router = (
    RunnablePassthrough()
    # 保留原始输入
    | RunnableLambda(lambda x: {"classify": classify_input(x["input"]), "raw": x["input"]})
    | RunnableBranch(
        (lambda x: x["classify"] == "physics", 
         lambda x: physics_chain.invoke({"input": x["raw"]})),  # 使用原始问题
        (lambda x: x["classify"] == "math", 
         lambda x: math_chain.invoke({"input": x["raw"]})),
        lambda x: general_chain.invoke({"input": x["raw"]})
    )
    # 后处理类（可选
    # | RunnableLambda(post_processor)
)

# 测试验证
if __name__ == "__main__":
    result1 = router.invoke({"input": "解释一下牛顿第一定律"})
    print(f"物理问题: {result1}")

    result2 = router.invoke({"input": "如何解二次方程？"})
    print(f"数学问题: {result2}")

    result3 = router.invoke({"input": "今天天气怎么样？"})
    print(f"通用问题: {result3}")

SequentalChain顺序执行

SequentialChain 是 LangChain 中用于顺序执行多个链的机制，前一个链的输出作为后一个链的输入，形成处理流水线

简单顺序执行

from langchain_core.runnables import RunnableSequence
from langchain_deepseek import ChatDeepSeek
from langchain.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from dotenv import load_dotenv
load_dotenv()

from langchain.globals import set_debug
set_debug(True)

template = """你是一个剧作家，我会给出一个剧的辩题，你的任务是为该标题写一个剧情概要
标题：{title}
剧作家："""

prompt1 = PromptTemplate(input_variables=["title"],template=template)
first_chain = prompt1 | ChatDeepSeek(model="deepseek-chat") | StrOutputParser()

template = """你是剧评人。给出剧本的摘要，你需要为该剧撰写评论：
剧本概述{synopsis}
剧评人：
"""
prompt2 = PromptTemplate(input_variables=["synopsis"],template=template)
secend_chain = prompt2 | ChatDeepSeek(model="deepseek-chat") | StrOutputParser()

# 运行两个chain
sequential_chain = first_chain | secend_chain # LCEL风格创建 
result = sequential_chain.invoke("赘婿龙王归来") 
print(result)

TransformationChain

TransformationChain（转换链）是LangChain中用于数据转换和预处理的专用链。它的核心作用是：原始数据 → 转换处理 → 结构化/标准化数据 → 供后续AI链使用

数据清洗：去除噪声、标准化格式
格式转换：文本→字典、JSON→文本等
特征提取：从原始数据中提取关键信息
数据增强：扩展或转换数据格式

其实就是自己写方法，进行数据的处理，然后集成到chain中

Chain摘要

Chain摘要主要用于对长文本进行内容总结和浓缩。

它的核心作用是将复杂的文档或对话内容提炼成简洁、关键的要点，以便于后续处理或人类快速理解。

stuff document

最简单，一次性处理所有文档，适用短文档

from langchain.chains.summarize import load_summarize_chain
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.docstore.document import Document
from langchain_deepseek import ChatDeepSeek
from dotenv import load_dotenv
load_dotenv()
# from langchain.globals import set_debug
# set_debug(True)

# 初始化模型
llm = ChatDeepSeek(model="deepseek-chat")

# 创建长文档（模拟长文本）
long_text = """
近年来，人工智能技术快速发展，在各个领域都有广泛应用。
在自然语言处理方面，深度学习模型如Transformer架构的出现极大地提升了机器理解人类语言的能力。
GPT系列模型展示了强大的文本生成和理解能力，能够完成写作、翻译、问答等多种任务。
在计算机视觉领域，卷积神经网络的发展使得图像识别、目标检测等任务达到了实用水平。
人工智能在医疗、金融、教育等行业也发挥着重要作用，帮助提高效率和准确性。
不过，人工智能的发展也带来了一些挑战，如数据隐私、算法偏见和就业影响等问题需要解决。
未来，人工智能技术将继续发展，与人类社会的融合将更加深入。
"""

# 分割文档
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
docs = text_splitter.create_documents([long_text])

# 使用StuffDocumentsChain
chain = load_summarize_chain(
    llm=llm,
    chain_type="stuff",  # 使用stuff方法
    verbose=True  # 显示详细过程
)

# 执行摘要
summary = chain.invoke(docs)
print("摘要结果:")
print(summary)
# 近年来，人工智能（AI）在自然语言处理（如Transformer、GPT系列）和计算机视觉（如卷积神经网络）等领域取得突破，广泛应用于医疗、金融、教育等行业，提升了效率与准确性。然而，AI也带来了数据隐私、算法偏见和就业影响等挑战。未来，AI技术将持续发展，并更深入地融入社会。

Map-Reduce

先分块处理再合并，适用长文档

# 使用MapReduce
chain = load_summarize_chain(
    llm=llm,
    chain_type="map_reduce", # 使用MapReduce
    verbose=True  # 显示详细过程
)
summary = chain.invoke(docs)
print("摘要结果:")
print(summary)
# 人工智能技术近年来快速发展并广泛应用，深度学习显著提升了自然语言处理和计算机视觉等领域的性能，已在医疗、金融等行业提高效率，但同时也面临数据隐私、算法偏见等挑战，未来将继续深化与社会融合。