大模型学习笔记：LangChain‘s Chain

Chain

Chain链：将组件串联，上一个组件的输出作为下一个组件的输入是LangChain链（尤其是|管道链）的核心工作原理，实现数据的自动化流转与组件的协同工作：

chain = prompt_template | model

前提：Runnable子类对象才能入链（以及Callable、Mapping接口子类对象也可以加入）。

重要点：

• 可以通过“|”符号来让各个组件行成链
• 成链的各个组件，需是Runnable接口的子类
• 形成的链是RunnableSerializable对象（Runnable接口子类）
• 可通过链调用invoke或stream触发真个链条的执行

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


chat_prompt_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一位与人交往风趣幽默的同事，情商超高。"),
        MessagesPlaceholder(variable_name="history"),
        ("human", "有个同事心情不好，你帮我安慰他")
    ]
)

history_data = [
    ("system", "你是一位与人交往风趣幽默的同事，情商超高。"),
    ("human", "有个同事心情不好，你帮我安慰他"),
    ("ai", "哦，发生什么事了？要不要跟我说说？"),
    ("human", "他被老板批评了"),
    ("ai", "被老板批评了确实挺难受的，不过没关系，大家都会有被批评的时候。重要的是从中吸取经验，下次做得更好！")
]

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

#组成链，每一个组件都是Runnable的子类，链式调用
chain = chat_prompt_template | model

#通过链调用invoke方法，传入输入
#res = chain.invoke(input={"history": history_data})
#print(res.content, type(res))

for chunk in chain.stream(input={"history": history_data}):
    print(chunk.content, end="", flush=True)

chain变量是RunnableSequence（RunnableSerializable 子类）类型。基于的对Runnable基类内部对__or__魔术方法的改写。后续继续使用|添加新的组件，依旧会得到RunnableSequence，这是链的基础架构。

StrOutputParser字符串输出解析器

StrOutputParser 是Langchain内置的简单字符串解析器：

• 可以将AIMessage解析为简单的字符串，符合了模型invoke方法要求（可传入字符串，不接受AIMessage类型）
• 是Runnable接口的子类，可加入链

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

model = ChatTongyi(model="qwen3-max")
prompt = PromptTemplate.from_template("我邻居姓{lastname}，刚生了{gender}，你帮我起个名字,别废话")
parser = StrOutputParser()

chain = prompt | model | parser | model | parser

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

JsonOutputParser

在构建链的时候注意整体兼容性，注意前后组件的输入和输出要求。

• 模型输入：PromptValue或字符串或序列（BaseMessage，list，tuple,str，dict）
• 模型输出：AIMessage
• 提示词模板输入：要求是字典
• 提示词模板输出：PromptValue对象
• StrOutputParser：AIMessage输入，str输出
• JsonOutputParser：AIMessage输入，dict输出

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")

#第一个提示词模板
first_prompt = PromptTemplate.from_template("我邻居姓{lastname}，刚生了{gender}，你帮我起个名字,并封装为json返回给我，要求key是name, value就是名字，请严格遵守")

#第二个提示词模板
second_prompt = PromptTemplate.from_template("姓名{name}, 请帮我解析含义。")

#构建链
chain = first_prompt | model | json_parser | second_prompt | model | str_parser

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

RunnableLambda

除了JsonOutputParser这类固定的解析器外，我们可以自己编写lambda匿名函数来完成自定义逻辑的数据转换，自由。RunnableLambda 类是LangChain内置的，将普通函数等转换为Runnable接口实例，方便自定义函数加入chain。

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

#创建解析器
str_parser = StrOutputParser()

#模型创建
model = ChatTongyi(model="qwen3-max")

#第一个提示词模板
first_prompt = PromptTemplate.from_template("我邻居姓{lastname}，刚生了{gender}，你帮我起个名字,仅告诉我名字，别废话")

#第二个提示词模板
second_prompt = PromptTemplate.from_template("姓名{name}, 请帮我解析含义。")

#传入的形参:AIMessage --> dict ({"name": "xxx"})
myFunc = RunnableLambda(lambda ai_msg: {"name": ai_msg.content})

#构建链
#chain = first_prompt | model | myFunc | second_prompt | model | str_parser
chain = first_prompt | model | (lambda ai_msg: {"name": ai_msg.content})  | second_prompt | model | str_parser

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