Langchain学习笔记(十一)：Chain构建与组合技巧

注：本文是Langchain框架的学习笔记；不是教程！不是教程！内容可能有所疏漏，欢迎交流指正。后续将持续更新学习笔记，分享我的学习心得和实践经验。

前言

在LangChain的发展过程中，API设计经历了重要的演进。从0.1.17版本开始，传统的Chain类（如LLMChain、SequentialChain等）被标记为已弃用，官方推荐使用更现代的LCEL（LangChain Expression Language）和管道符（|）语法。本文将详细对比新旧两种方式，帮助大家理解这一重要变化

1. Chain的基本概念与版本变化

1.1 什么是Chain？

Chain（链）是LangChain中用于连接多个组件的核心抽象。想象一下工厂的流水线，每个工作站负责一个特定的任务，产品从一个工作站传递到下一个工作站，最终完成整个生产过程。Chain就是这样的概念：

输入处理：接收用户的原始输入
中间处理：通过多个步骤对数据进行转换和处理
输出生成：产生最终的结果

Chain的核心价值在于：

模块化设计：将复杂的AI任务分解为多个简单的步骤
可重用性：每个组件都可以在不同的场景中重复使用
可维护性：便于调试、测试和优化单个环节
灵活组合：可以根据需求灵活组合不同的处理步骤

1.2 版本演进的背景

为什么要从传统Chain迁移到LCEL？

传统的Chain类虽然功能强大，但存在一些局限性：

性能瓶颈：无法充分利用并行处理能力
语法复杂：需要创建多个类实例，代码冗长
功能限制：不支持流式处理和高级组合模式
维护困难：复杂的继承关系增加了维护成本

LCEL（LangChain Expression Language）的优势：

直观语法：使用管道符（|）连接组件，类似Unix管道
高性能：原生支持并行、批处理和流式处理
类型安全：更好的类型提示和错误检查
易于调试：清晰的数据流向，便于问题定位

1.3 新旧版本对比表

功能	旧版本（已弃用）	新版本（推荐）	说明
基础链	LLMChain	prompt \| llm	最基础的提示词+模型组合
顺序链	SequentialChain	chain1 \| chain2 \| chain3	按顺序执行多个处理步骤
执行方法	chain.run()	chain.invoke()	单次执行链的方法
批量执行	chain.apply()	chain.batch()	批量处理多个输入
异步执行	chain.arun()	chain.ainvoke()	异步执行，提高并发性能
流式执行	不支持	chain.stream()	实时流式输出，改善用户体验

2. LLMChain的新旧版本对比

2.1 LLMChain的作用和意义

LLMChain是LangChain中最基础、最常用的链类型。它的作用是将提示词模板（Prompt Template）与大语言模型（LLM）连接起来，形成一个完整的处理单元。

LLMChain的核心功能：

模板化输入：使用提示词模板，支持变量替换
模型调用：自动调用指定的大语言模型
输出处理：可选的输出解析和格式化
错误处理：统一的错误处理机制

使用场景：

文本生成（如写作助手、内容创作）
问答系统（基于提示词的简单问答）
文本分析（情感分析、关键词提取等）
格式转换（如JSON生成、代码生成）

2.2 基础用法对比

旧版本写法（已弃用）

from langchain_ollama import ChatOllama
from langchain.chains import LLMChain  # 已弃用
from langchain.prompts import PromptTemplate

llm = ChatOllama(
    base_url=OLLAMA_BASE_URL,
    model=MODEL_NAME,
    temperature=0.1  # 降低温度以获得更确定性的回答
)

prompt_template = PromptTemplate(
    input_variables=["topic"],
    template="请为我写一篇关于{topic}的简短介绍，大约100字左右。"
)

# 构建LLMChain（已弃用）
llm_chain = LLMChain(
    llm=llm,
    prompt=prompt_template,
    verbose=True
)

# 使用链（已弃用的方法）
result = llm_chain.run(topic="人工智能")  # 弃用警告
print(result)

新版本写法（推荐）

from langchain_core.prompts import PromptTemplate
from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser

OLLAMA_BASE_URL = "http://192.168.0.118:11434"
MODEL_NAME = "deepseek-r1:1.5b"

llm = ChatOllama(
    base_url=OLLAMA_BASE_URL,
    model=MODEL_NAME,
    temperature=0.1  # 降低温度以获得更确定性的回答
)

prompt_template = PromptTemplate(
    input_variables=["topic"],
    template="请为我写一篇关于{topic}的简短介绍，大约100字左右。"
)

# 使用管道符构建链（推荐）
chain = prompt_template | llm | StrOutputParser()

# 使用invoke方法执行（推荐）
result = chain.invoke({"topic": "人工智能"})
print(result)

# 新版本支持的其他执行方式
# 批量执行
results = chain.batch([
    {"topic": "人工智能"},
    {"topic": "机器学习"}
])
print(results)

新版本的优势解析：

管道符语法：prompt_template | llm | StrOutputParser() 清晰地表达了数据流向
输出解析器：StrOutputParser() 确保输出为字符串格式
批量处理：batch() 方法可以高效处理多个输入
类型安全：更好的类型提示，减少运行时错误

2.3 带输出解析器的高级用法

输出解析器的作用：

输出解析器（Output Parser）是LangChain中用于处理模型输出的重要组件。它的主要作用包括：

格式化输出：将模型的原始输出转换为结构化数据
类型转换：确保输出符合预期的数据类型
验证检查：验证输出是否符合预定义的格式要求
错误处理：处理格式不正确的输出

新版本写法

from langchain_core.prompts import PromptTemplate
from langchain_ollama import ChatOllama
from langchain.output_parsers import CommaSeparatedListOutputParser

OLLAMA_BASE_URL = "http://192.168.0.118:11434"
MODEL_NAME = "deepseek-r1:1.5b"

llm = ChatOllama(
    base_url=OLLAMA_BASE_URL,
    model=MODEL_NAME,
    temperature=0.1  # 降低温度以获得更确定性的回答
)

# 新版本
output_parser = CommaSeparatedListOutputParser()

list_prompt = PromptTemplate(
    template="列出5个关于{topic}的关键词，用逗号分隔。\n{format_instructions}",
    input_variables=["topic"],
    partial_variables={"format_instructions": output_parser.get_format_instructions()}
)

# 使用管道符连接解析器
chain = list_prompt | llm | output_parser

keywords = chain.invoke({"topic": "机器学习"})
print(f"关键词列表: {keywords}")
print(f"类型: {type(keywords)}")

3. SequentialChain

3.1 SequentialChain的概念和作用

SequentialChain（顺序链）是用于按顺序执行多个处理步骤的链类型。它的核心思想是将复杂的任务分解为多个简单的子任务，然后按照特定的顺序依次执行。

SequentialChain的特点：

步骤化处理：将复杂任务分解为多个简单步骤
数据传递：前一个步骤的输出作为后一个步骤的输入
模块化设计：每个步骤都是独立的，便于测试和维护
灵活组合：可以根据需要调整步骤的顺序和组合

适用场景：

内容创作流程：大纲生成 → 内容写作 → 格式优化
数据分析流程：数据收集 → 数据清洗 → 分析报告
产品开发流程：需求分析 → 方案设计 → 实施计划
客户服务流程：问题识别 → 解决方案 → 回复生成

3.2 简单顺序链

简单顺序链的特点：

每个步骤只有一个输入和一个输出
前一步的输出直接作为后一步的输入
适合线性处理流程

新版本写法

from langchain_core.prompts import PromptTemplate
from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser

OLLAMA_BASE_URL = "http://192.168.0.118:11434"
MODEL_NAME = "deepseek-r1:1.5b"

llm = ChatOllama(
    base_url=OLLAMA_BASE_URL,
    model=MODEL_NAME,
    temperature=0.1  # 降低温度以获得更确定性的回答
)

# 新版本
outline_prompt = PromptTemplate(
    input_variables=["topic"],
    template="为主题'{topic}'创建一个详细的文章大纲，包含3-5个主要部分。"
)

article_prompt = PromptTemplate(
    input_variables=["outline"],
    template="根据以下大纲写一篇简短的文章：\n{outline}"
)

# 使用管道符构建顺序链
writing_chain = (
        outline_prompt
        | llm
        | StrOutputParser()
        | (lambda outline: {"outline": outline})
        | article_prompt
        | llm
        | StrOutputParser()
)

final_article = writing_chain.invoke({"topic": "可持续发展"})
print(final_article)

3.2 复杂顺序链

复杂顺序链的特点：

支持多个输入和输出变量
可以在处理过程中保留和传递多个数据
支持并行处理某些步骤
适合复杂的业务逻辑

新版写法

from langchain_core.prompts import PromptTemplate
from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableParallel

OLLAMA_BASE_URL = "http://192.168.0.118:11434"
MODEL_NAME = "deepseek-r1:1.5b"

llm = ChatOllama(
    base_url=OLLAMA_BASE_URL,
    model=MODEL_NAME,
    temperature=0.1  # 降低温度以获得更确定性的回答
)

# 新版本
analysis_prompt = PromptTemplate(
    input_variables=["product_name", "product_description"],
    template="产品：{product_name}\n描述：{product_description}\n请分析产品特点："
)

slogan_prompt = PromptTemplate(
    input_variables=["product_name", "product_analysis"],
    template="产品：{product_name}\n分析：{product_analysis}\n请创建3个营销标语："
)

# 使用RunnableParallel和管道符构建复杂链
product_marketing_chain = (
        RunnablePassthrough.assign(product_analysis=analysis_prompt | llm | StrOutputParser()) | RunnableParallel(
    product_analysis=lambda x: x["product_analysis"], marketing_slogans=slogan_prompt | llm | StrOutputParser()
)
)

result = product_marketing_chain.invoke({
    "product_name": "智能语音助手",
    "product_description": "基于AI技术的智能语音助手"
})

print("产品分析：", result["product_analysis"])
print("营销标语：", result["marketing_slogans"])

核心组件解析：

RunnablePassthrough.assign()：
- 作用：在保留原有输入的基础上，添加新的字段
- 这里添加了 product_analysis 字段，包含产品分析结果
RunnableParallel()：
- 作用：并行执行多个任务，提高处理效率
- 同时输出产品分析和营销标语两个结果

处理流程：

输入处理：接收产品名称和描述
产品分析：生成产品特点分析，并保留原始输入
并行输出：同时生成最终的分析结果和营销标语

4. RouterChain

4.1 RouterChain的概念和作用

RouterChain（路由链）是一种智能分发机制，它可以根据输入内容的特征，自动选择最合适的处理路径。就像交通路口的智能信号灯，根据不同方向的车流量来调整信号，RouterChain根据不同类型的输入来选择最佳的处理方式。

RouterChain的核心价值：

智能分类：自动识别输入内容的类型或意图
专业处理：为不同类型的问题提供专门的处理逻辑
提高效率：避免用通用方法处理所有问题
提升质量：专业化处理通常能获得更好的结果

典型应用场景：

智能客服系统：根据问题类型路由到不同的处理模块
内容分类处理：技术问题、商业问题、创意问题分别处理
多语言支持：根据语言类型选择对应的处理器
难度分级：简单问题快速回答，复杂问题深度分析

4.2 基础路由链

路由逻辑的设计思路：

关键词匹配：通过关键词识别问题类型
专业模板：为每种类型设计专门的提示词模板
默认处理：为无法分类的问题提供通用处理方式

新版写法

from langchain_core.prompts import PromptTemplate
from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableBranch

OLLAMA_BASE_URL = "http://192.168.0.118:11434"
MODEL_NAME = "deepseek-r1:1.5b"

llm = ChatOllama(
    base_url=OLLAMA_BASE_URL,
    model=MODEL_NAME,
    temperature=0.1  # 降低温度以获得更确定性的回答
)

# 新版本
tech_prompt = PromptTemplate(
    template="你是技术专家。请回答：{input}",
    input_variables=["input"]
)

business_prompt = PromptTemplate(
    template="你是商业顾问。请回答：{input}",
    input_variables=["input"]
)

default_prompt = PromptTemplate(
    template="请回答：{input}",
    input_variables=["input"]
)


# 使用RunnableBranch实现路由逻辑
def route_question(x):
    question = x["input"].lower()
    if any(keyword in question for keyword in ["代码", "编程", "技术", "开发"]):
        return "tech"
    elif any(keyword in question for keyword in ["商业", "市场", "销售", "管理"]):
        return "business"
    else:
        return "default"


router_chain = RunnableBranch(
    (lambda x: route_question(x) == "tech", tech_prompt | llm | StrOutputParser()),
    (lambda x: route_question(x) == "business", business_prompt | llm | StrOutputParser()),
    default_prompt | llm | StrOutputParser()
)

result = router_chain.invoke({"input": "如何优化Python代码性能？"})
print(result)

核心组件详解：

route_question() 函数：
- 作用：分析输入问题，判断问题类型
- 逻辑：通过关键词匹配来识别问题领域
- 返回：问题类型标识（tech/business/default）
RunnableBranch：
- 作用：根据条件选择不同的处理分支
- 结构：(条件函数, 处理链) 的元组列表
- 默认：最后一个参数作为默认处理分支
专业化提示词：
- tech_prompt：以技术专家身份回答技术问题
- business_prompt：以商业顾问身份回答商业问题
- default_prompt：通用回答模式