LangChain 进阶：Chain 链与输出解析器

系列文章：RAG 与 LangChain 开发实战（3/7）
阅读时间：约 25 分钟
前置知识：LangChain 框架入门

一、什么是 Chain 链？

Chain（链） 是 LangChain 的核心概念，指将多个组件串联在一起，按顺序执行，前一个组件的输出作为下一个组件的输入。

┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│  Prompt     │ -> │   Model     │ -> │   Parser    │
│  Template   │    │             │    │             │
└─────────────┘    └─────────────┘    └─────────────┘

核心价值：实现数据的自动化流转与组件协同工作。

二、| 运算符与链式调用

2.1 基础链式调用

LangChain 使用 | 运算符实现链式调用：

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.chat_models.tongyi import ChatTongyi

chat_prompt_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个边塞诗人。"),
        MessagesPlaceholder("history"),
        ("human", "请再来一首唐诗"),
    ]
)

history_data = [
    ("human", "写一首唐诗"),
    ("ai", "床前明月光，疑是地上霜"),
]

model = ChatTongyi(model="qwen3-max")

# 组成链 - 每个组件都是 Runnable 接口子类
chain = chat_prompt_template | model

# 流式输出
for chunk in chain.stream({"history": history_data}):
    print(chunk.content, end="", flush=True)

2.2 | 运算符的本质

| 运算符是 Python 魔法方法 __or__ 的重写：

# Python 运算符与魔法方法对应
a + b  # 调用 a.__add__(b)
a | b  # 调用 a.__or__(b)

LangChain 的 Runnable 基类重写了 __or__ 方法，使组件可以链式组合。

自定义示例：

class Test:
    def __init__(self, name):
        self.name = name

    def __or__(self, other):
        return MySequence(self, other)

    def __str__(self):
        return self.name


class MySequence:
    def __init__(self, *args):
        self.sequence = []
        for arg in args:
            self.sequence.append(arg)

    def __or__(self, other):
        self.sequence.append(other)
        return self

    def run(self):
        for i in self.sequence:
            print(i)


# 使用
a = Test('a')
b = Test('b')
c = Test('c')
d = a | b | c
d.run()  # 输出：a b c

2.3 Runnable 接口

所有可入链的组件都实现 Runnable 接口：

from langchain_core.prompts import PromptTemplate
from langchain_community.llms.tongyi import Tongyi

prompt = PromptTemplate.from_template("你是一个 AI 助手")
model = Tongyi(model="qwen3-max")

# 链可以是任意长度
chain = prompt | model | prompt | model

print(type(chain))
# <class 'langchain_core.runnables.base.RunnableSequence'>

# 链本身也是 Runnable，支持 invoke/stream
chain.invoke({...})
chain.stream({...})

关键前提：只有 Runnable 子类对象才能入链（以及 Callable、Mapping 接口子类）。

三、输出解析器（Output Parser）

3.1 为什么需要解析器？

考虑以下场景：想用第一个模型的输出作为第二个模型的输入。

from langchain_core.prompts import PromptTemplate
from langchain_community.llms.tongyi import Tongyi

prompt_template = PromptTemplate.from_template(
    "邻居姓{lastname}，生了{gender}，起个名字。"
)
model = Tongyi(model="qwen-max")

# 错误示范 - 直接连接两个模型
chain = prompt_template | model | model  # ❌ 会报错！

res = chain.invoke({"lastname": "张", "gender": "女儿"})

错误原因：

  • 第一个 model 输出：AIMessage 类型
  • 第二个 model 输入要求：PromptValue 或 str
  • 类型不匹配！

3.2 StrOutputParser - 字符串解析器

将 AIMessage 转换为普通字符串：

from langchain_core.output_parsers import StrOutputParser

parser = StrOutputParser()

# 正确写法
chain = prompt_template | model | parser | model

res = chain.invoke({"lastname": "张", "gender": "女儿"})
print(res)

数据流：

PromptTemplate → PromptValue
      ↓
      model → AIMessage
      ↓
   parser → str
      ↓
      model → AIMessage

3.3 JsonOutputParser - JSON 解析器

将模型输出的 JSON 字符串解析为字典：

from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import PromptTemplate

str_parser = StrOutputParser()
json_parser = JsonOutputParser()

model = ChatTongyi(model="qwen3-max")

# 第一个提示词 - 要求返回 JSON
first_prompt = PromptTemplate.from_template(
    "邻居姓{lastname}，生了{gender}，起个名字，"
    "封装为 JSON 格式返回。key 是 name，value 是名字。"
)

# 第二个提示词 - 解析名字含义
second_prompt = PromptTemplate.from_template(
    "姓名：{name}，请解析含义。"
)

# 构建链
chain = (
    first_prompt 
    | model 
    | json_parser      # AIMessage → dict
    | second_prompt 
    | model 
    | str_parser       # AIMessage → str
)

for chunk in chain.stream({"lastname": "张", "gender": "女儿"}):
    print(chunk, end="", flush=True)

3.4 组件输入输出对照表

组件 输入 输出
模型 PromptValue / str AIMessage
Prompt 模板 dict PromptValue
StrOutputParser AIMessage str
JsonOutputParser AIMessage dict

四、RunnableLambda - 自定义转换

除了内置解析器，还可以用 RunnableLambda 实现自定义逻辑。

4.1 基本用法

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_community.chat_models.tongyi import ChatTongyi

model = ChatTongyi(model="qwen3-max")
str_parser = StrOutputParser()

first_prompt = PromptTemplate.from_template(
    "邻居姓{lastname}，生了{gender}，起个名字，只返回名字。"
)

second_prompt = PromptTemplate.from_template(
    "姓名{name}，请解析含义。"
)

# 方式 1：显式使用 RunnableLambda
from langchain_core.runnables import RunnableLambda
chain = (
    first_prompt 
    | model 
    | RunnableLambda(lambda ai_msg: {"name": ai_msg.content}) 
    | second_prompt 
    | model 
    | str_parser
)

# 方式 2：直接使用 lambda 函数（推荐）
# Runnable 的__or__支持 Callable 接口
chain = (
    first_prompt 
    | model 
    | (lambda ai_msg: {"name": ai_msg.content})  # 更简洁
    | second_prompt 
    | model 
    | str_parser
)

for chunk in chain.stream({"lastname": "曹", "gender": "女孩"}):
    print(chunk, end="", flush=True)

4.2 数据转换流程

AIMessage("{name: 张若曦}")
        ↓
lambda ai_msg: {"name": ai_msg.content}
        ↓
{"name": "张若曦"}  ← 字典可注入 Prompt 模板

五、多模型链实战

5.1 完整示例：起名 + 解析

from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import PromptTemplate

model = ChatTongyi(model="qwen3-max")

# 第一步：生成名字（JSON 格式）
first_prompt = PromptTemplate.from_template(
    """
    邻居姓{lastname}，生了{gender}，请帮忙起名字。
    要求：
    1. 仅生成一个名字
    2. 以 JSON 格式返回：{{"name": "名字"}}
    3. 不要额外信息
    """
)

# 第二步：解析名字含义
second_prompt = PromptTemplate.from_template(
    """
    姓名：{name}
    请帮我解析这个名字的含义和寓意。
    """
)

# 构建完整链条
chain = (
    first_prompt           # 输入：{lastname, gender} → 输出：PromptValue
    | model                # 输入：PromptValue → 输出：AIMessage("{name: xxx}")
    | JsonOutputParser()   # 输入：AIMessage → 输出：{"name": "xxx"}
    | second_prompt        # 输入：dict → 输出：PromptValue
    | model                # 输入：PromptValue → 输出：AIMessage
    | StrOutputParser()    # 输入：AIMessage → 输出：str
)

# 执行
for chunk in chain.stream({"lastname": "李", "gender": "男孩"}):
    print(chunk, end="", flush=True)

5.2 链式调用最佳实践

  1. 明确每个组件的输入输出类型
  2. 使用解析器确保类型匹配
  3. 复杂转换用 RunnableLambda
  4. 链本身也是 Runnable，可继续组合

六、关键知识点总结

6.1 核心概念

概念 说明
Chain 组件串联，前一个输出作为下一个输入
| 运算符 __or__ 魔法方法重写，实现链式调用
Runnable 所有可入链组件的基接口
StrOutputParser AIMessage → str
JsonOutputParser AIMessage → dict
RunnableLambda 自定义转换逻辑

6.2 链构建规则

合法链：
prompt | model | parser | model

非法链（类型不匹配）：
model | model  # AIMessage 不能直接输入 model

6.3 代码速查

from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.runnables import RunnableLambda

# 字符串解析
parser = StrOutputParser()

# JSON 解析
json_parser = JsonOutputParser()

# 自定义转换
lambda ai_msg: {"name": ai_msg.content}

# 链式调用
chain = prompt | model | parser | model

七、下一步

本文介绍了 Chain 链和输出解析器。但实际应用还需要：

  • 会话记忆：让模型记住历史对话
  • 文档处理：加载和分割大型文档
  • 向量存储：实现语义检索

下一篇我们将讲解 LangChain 的记忆功能 和 文档处理组件。

系列文章导航

  1. LLM 基础与提示词工程：从 API 调用到 Prompt 优化技巧
  2. LangChain 框架入门：模型抽象与 Prompt 模板
  3. ✅ LangChain 进阶：Chain 链与输出解析器（本文）
  4. LangChain 记忆与文档处理
  5. RAG 核心概念与向量存储
  6. RAG 实战 (上)：知识库构建
  7. RAG 实战 (下)：智能客服系统

版权声明：本文基于学习笔记整理，转载请注明出处。

# python # 人工智能
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

当 AI Agent 拿到了你的终端和内网：一份 Agent 安全与沙箱化工程地图

avatar AI Agent技术社区
cover

AI时代，Android程序员何去何从？

avatar AI Agent技术社区
cover

你的工作是「写循环」：一文读懂 2026 年最新范式 Loop Engineering

avatar AI Agent技术社区