LangChain 案例实战：顺序链、分支链、循环链

作者：AI时代的一滴水

2025-04-21
北京
本文字数：4262 字
阅读完需：约 14 分钟

在 LangChain 应用中，单独使用 LLM 是可以的，但更复杂的应用需要将 LLM 进行链接 - 要么相互链接，要么与其他组件链接。

LangChain 为这种"链接"应用提供了 Chain 接口。LangChain 将链定义得非常通用，它是对组件调用的序列，可以包含其他链。

将组件组合成链的思想简单而强大。它极大地简化了复杂应用的实现，并使应用更模块化，从而更容易调试、维护和改进应用。

本文将通过三个渐进式案例展示如何使用 LangChain 的 Chain 构建高效的大语言模型工作流。我们将从简单的链式组合开始，逐步深入到路由逻辑和循环链的复杂场景。

1. 顺序链：链式组合（Sequential Chains）

本质：函数式编程的 compose

操作核心：数据流单向管道（A→B→C）

一句话介绍："LLM 版的 UNIX 管道"

案例：文本总结+翻译组合链

from langchain_core.prompts import ChatPromptTemplatefrom langchain_openai import ChatOpenAI
# 初始化模型llm = ChatOpenAI(base_url="https://api.deepseek.com/",                 api_key="your-api-key",                model="deepseek-chat")
# 1. 总结链summary_chain = (    ChatPromptTemplate.from_template("用一句话总结：{text}")    | llm)
# 2. 翻译链translation_chain = (    ChatPromptTemplate.from_template("翻译成中文：{text}")    | llm)
# 组合链（管道操作）report_chain = summary_chain | translation_chain
# 执行english_text = "LangChain provides tools to build LLM-powered applications."result = report_chain.invoke({"text": english_text})print(result.content)

复制代码

输出如下

# python e./combine_chain.pyLangChain 提供了用于开发由大型语言模型（LLMs）驱动的应用程序的工具。

复制代码

关键点：

使用|管道符连接多个链
前一个链的输出自动作为下一个链的输入
适合线性任务流（如：总结→翻译→格式化）

2. 分支链：路由链（Router Chains）

本质：模式匹配的状态机

核心：(输入→谓词)→目标链的映射

一句话介绍："LLM 流量控制器"

案例：领域专家路由系统

from langchain_core.prompts import ChatPromptTemplatefrom langchain_core.runnables import RunnableBranchfrom langchain_openai import ChatOpenAI
api_key = "your-api-key"api_base = "https://api.deepseek.com/"
# 1. 定义子链physics_chain = (    ChatPromptTemplate.from_template("你是一个物理学家，请专业回答：{input}")    | ChatOpenAI(base_url=api_base, api_key=api_key, model="deepseek-chat"))
math_chain = (    ChatPromptTemplate.from_template("你是一个数学家，请分步骤解答：{input}")    | ChatOpenAI(base_url=api_base, api_key=api_key, model="deepseek-chat"))
general_chain = (    ChatPromptTemplate.from_template("回答这个问题：{input}")    | ChatOpenAI(base_url=api_base, api_key=api_key, model="deepseek-chat"))
# 2. 构建路由链 (使用 RunnableBranch)router_chain = RunnableBranch(    (lambda x: "物理" in x["input"], physics_chain),    (lambda x: "数学" in x["input"], math_chain),    general_chain)
if __name__ == "__main__":    # 测试输入    test_input = {"input": "解释相对论"}    print(router_chain.invoke(test_input).content)  # 路由到物理链
    test_input = {"input": "解方程x^2+2x+1=0"}    print(router_chain.invoke(test_input).content)  # 路由到数学链

复制代码

输出如下

关键点：

RunnableBranch实现条件路由
每个分支可定制不同的 prompt 模板
适合多场景分流（如客服、教育领域）

3. 循环链

3.1 代码生成与验证链 V1

让 chain 执行自定义接口

本质：编译器前端简化版

核心：生成→解析→沙箱三级流水线

一句话介绍："AI 时代的编译器"

案例：AI 代码生成+语法检查

from langchain_core.runnables import RunnableLambdaimport ast, re
api_key = "your-api-key"api_base = "https://api.deepseek.com/"
# 1. 代码生成链code_chain = (    ChatPromptTemplate.from_template("用Python实现：{task}（只返回代码内容部分）")    | ChatOpenAI(base_url=api_base, api_key=api_key, model="deepseek-chat")    | {"code": RunnablePassthrough()}  # 保留原始输出)
# 2. 语法检查链def syntax_check(code_dict):    try:        print(f"正在检查代码语法...")        # 从 AIMessage 中提取代码内容        content = code_dict["code"].content        code_match = re.search(r'```python\n(.*?)\n```', content, re.DOTALL)        if code_match:            code = code_match.group(1)            ast.parse(code)            return {**code_dict, "status": "VALID", "code": code}        return {"status": "INVALID: No code block"}    except SyntaxError as e:        return {**code_dict, "status": f"INVALID: {e}"}
# 3. 组合循环链full_chain = code_chain | RunnableLambda(syntax_check)
if __name__ == "__main__":    # 测试输入    test_input = {"task": "编写快速排序函数"}    result = full_chain.invoke(test_input)    print("\n=== 语法检查结果 ===")    print(result["status"])    if result["status"] == "VALID":        print("\n=== 生成的代码 ===")        print(result["code"])

复制代码

结果输出

# python ./code_chain_v1.pyVALID                                                                                                                                     {}})}
=== 生成的代码 ===def quick_sort(arr):    if len(arr) <= 1:        return arr    pivot = arr[len(arr) // 2]    left = [x for x in arr if x < pivot]    middle = [x for x in arr if x == pivot]    right = [x for x in arr if x > pivot]    return quick_sort(left) + middle + quick_sort(right)

复制代码

关键点：

RunnableLambda实现自定义处理逻辑
正则表达式提取 Markdown 代码块
ast.parse进行 Python 语法验证
适合代码生成类场景的完整流水线

3.1 代码生成与验证链 V2: 循环执行

import refrom typing import Any, Dictfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.runnables import RunnableLambda, RunnablePassthroughfrom langchain_openai import ChatOpenAIimport ast
api_key = "your-api-key"api_base = "https://api.deepseek.com/"
# 1. 代码生成链code_chain = (    ChatPromptTemplate.from_template("用Python实现：{task}，只返回代码")    | ChatOpenAI(base_url=api_base, api_key=api_key, model="deepseek-chat")    | RunnablePassthrough()  # 保留原始输出)
# 2. 语法检查链def syntax_check(code_dict: Dict[str, Any]) -> Dict[str, Any]:    error = None    print(code_dict)    # 从 AIMessage 中提取代码内容    code = code_dict["code"].content    try:        print(f"正在检查代码语法...")        code_match = re.search(r'```python\n(.*?)\n```', code, re.DOTALL)        if code_match:            code = code_match.group(1)            ast.parse(code)            return {**code_dict, "status": "VALID", "extracted_code": code}        return {"status": "INVALID: No code block"}    except SyntaxError as e:        error = f"语法错误：{e}"    except Exception as e:        error = f"解析错误：{e}"
    print(f"代码语法检查失败：{error}, code: {code}")    # 自动修复（将错误信息反馈给LLM）    task = code_dict["task"]    repair_prompt = ChatPromptTemplate.from_template(        "修复这段Python代码的语法错误：\n"        "原始任务：{task}\n"        "错误信息：{error}\n"        "代码：\n```python\n{code}\n```"    )    print(repair_prompt)    repair_chain = repair_prompt | ChatOpenAI(base_url=api_base, api_key=api_key, model="deepseek-chat")        repaired = repair_chain.invoke({        "task": code_dict["task"],        "error": error,        "code": code,    })        # 递归调用继续验证    return syntax_check({        **code_dict,        "code": repaired,    })
# 3. 组合链full_chain = {    "task": RunnablePassthrough(), ## 保留原始任务    "code": code_chain,} | RunnableLambda(syntax_check)
if __name__ == "__main__":    # 测试输入    test_input = "编写快速排序函数"    result = full_chain.invoke(test_input)    print("\n=== 语法检查结果 ===")    print(result["status"])    if result["status"] == "VALID":        print("\n=== 生成的代码 ===")        print(result["extracted_code"])

复制代码

循环输出

# python.exe ./code_chain_v2.py正在检查代码语法...
=== 语法检查结果 ===INVALID: invalid syntax (<unknown>, line 1)
正在检查代码语法...=== 语法检查结果 ===VALID
=== 生成的代码 ===def quick_sort(arr):    """    快速排序函数（升序）    :param arr: 待排序的列表    """    # 递归终止条件：当数组长度为0或1时，直接返回    if len(arr) <= 1:        return arr
    pivot = arr[len(arr) // 2]  # 选择中间元素作为基准值（也可以选第一个/最后一个）    left = [x for x in arr if x < pivot]    # 小于基准值的子数组    middle = [x for x in arr if x == pivot] # 等于基准值的子数组    right = [x for x in arr if x > pivot]   # 大于基准值的子数组
    return quick_sort(left) + middle + quick_sort(right)
# 示例用法if __name__ == "__main__":    test_array = [3, 6, 8, 10, 1, 2, 1]    print("原始数组:", test_array)    sorted_array = quick_sort(test_array)    print("排序后数组:", sorted_array)

复制代码

关键点：

RunnableLambda实现自定义处理逻辑
正则表达式提取 Markdown 代码块
ast.parse进行 Python 语法验证
当语法检查失败时，附带原始问题+错误代码+错误原因同时输入给 LLM，继续执行，直到代码检查通过，或者实现设置失败阈值
适合代码生成类场景的完整流水线

对比总结

LangChain 资料

LangChain官网

LangChain 架构图

发布于: 刚刚阅读数: 5

原文链接:【http://xie.infoq.cn/article/500d4c0d536816ad08e81b4ae】。

AI时代的一滴水

关注

路虽远，行则必至；事虽难，做则必成 2020-07-21 加入

还未添加个人简介

发布

暂无评论

创作场景

LangChain 案例实战：顺序链、分支链、循环链

1. 顺序链：链式组合（Sequential Chains）

案例：文本总结+翻译组合链

2. 分支链：路由链（Router Chains）

案例：领域专家路由系统

3. 循环链

3.1 代码生成与验证链 V1

案例：AI 代码生成+语法检查

3.1 代码生成与验证链 V2: 循环执行

对比总结

LangChain 资料

AI时代的一滴水

评论