企业官网建设流程全解析

一、链是什么

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

chain = prompt_template | model

1.1 谁能入链

核心前提：即 Runnable 子类对象才能入链（以及 Callable、Mapping 接口子类对象也可加入）。目前学到的组件均是 Runnable 接口的子类。

1.2 链执行起来什么样

通过|链接提示词模板对象和模型对象：

• 返回值 chain 是RunnableSerializable对象

• 它是 Runnable 接口的直接子类，也是绝大多数组件的父类

• 通过invoke或stream进行阻塞执行或流式执行

• 组成的链：上一个组件的输出作为下一个组件的输入

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_community.chat_models.tongyi import ChatTongyi from langchain_core.runnables.base import RunnableSerializable chat_prompt_template = ChatPromptTemplate.from_messages( [ ("system", "你是一个边塞诗人，可以作诗。"), MessagesPlaceholder("history"), ("human", "请再来一首唐诗，无需额外输出"), ] ) history_data = [ ("human", "你来写一个唐诗"), ("ai", "床前明月光，疑是地上霜，举头望明月，低头思故乡"), ("human", "好诗再来一个"), ("ai", "锄禾日当午，汗滴禾下锄，谁知盘中餐，粒粒皆辛苦"), ] model = ChatTongyi(model="qwen3-max") chain: RunnableSerializable = chat_prompt_template | model print(type(chain)) # Runnable接口，invoke执行 res = chain.invoke({"history": history_data}) print(res.content) # Runnable接口，stream执行 for chunk in chain.stream({"history": history_data}): print(chunk.content, end="", flush=True)

1.3 链小结

• 链是将各个组件串联在一起，按顺序执行，前一个组件的输出作为下一个组件的输入

• 通过|符号让各个组件形成链

• 成链的各个组件，需是 Runnable 接口的子类

• 形成的链是 RunnableSerializable 对象

• 可通过链调用invoke或stream触发整个链条的执行

二、|运算符为什么能组链

2.1 本质是__or__

chain = chat_prompt_template | model在语法上使用了|运算符的重写。

在 Python 中，运算符行为由类的魔法方法决定：

• a + b本质调用a.__add__(b)

• a | b本质调用a.__or__(b)

自行实现__or__，即可重写|。

2.2 课件示例：a | b | c

class Test(object): def __init__(self, name): self.name = name def __str__(self): return f"Test({self.name})" def __or__(self, other): return MySequence(self, other) class MySequence(object): 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 arg in self.sequence: print(arg) if __name__ == "__main__": a = Test("a") b = Test("b") c = Test("c") d = a | b | c d.run() print(type(d))

2.3 落到 LangChain 上

chain = prompt | model得到的是RunnableSequence（RunnableSerializable 子类），原因就是 Runnable 基类内部对__or__的改写。后面继续用|加组件，依旧得到 RunnableSequence——这就是链的基础架构。

三、StrOutputParser：为什么 prompt | model | model 会报错

3.1 复现

需求：第一次模型的输出，再拿去第二次询问模型。

from langchain_core.prompts import PromptTemplate from langchain_community.chat_models.tongyi import ChatTongyi model = ChatTongyi(model="qwen3-max") prompt = PromptTemplate.from_template( "我邻居姓：{lastname}, 刚生了{gender}，请起名，仅告知名字无需其它内容" ) chain = prompt | model | model res = chain.invoke({"lastname": "张", "gender": "女儿"}) print(res.content)

运行报错：

ValueError: Invalid input type <class 'langchain_core.messages.ai.AIMessage'>. Must be a PromptValue, str, or list of BaseMessages.

3.2 原因

• prompt 的结果是PromptValue，输入给了 model —— 这一段是合法的

• model 的输出是AIMessage

• 模型invoke的 input 类型是LanguageModelInput = PromptValue | str | Sequence[MessageLikeRepresentation]，不接收 AIMessage

3.3 用 StrOutputParser 做类型转换

LangChain 内置StrOutputParser字符串输出解析器：把 AIMessage 解析为简单字符串，且它是 Runnable 子类，可以加入链。

parser = StrOutputParser() chain = prompt | model | parser | model

小结：StrOutputParser 是内置的简单字符串解析器，可以将 AIMessage 转换为基础字符串，可以加入 chain。

四、JsonOutputParser 与标准多模型链

4.1 更标准的处理逻辑

prompt | model | parser | model并不标准：上一个模型的输出没有被处理成「下一个提示词模板」所需的输入。

正常逻辑：

invoke / stream 初始输入 → 提示词模板 → 模型 → 数据处理 → 提示词模板 → 模型 → 解析器 → 结果

即：上一个模型的输出，应作为提示词模板的输入，构建下一个提示词，用来二次调用模型。

• 模型输出：AIMessage

• 提示词模板invoke要求输入：dict，输出：PromptValue

• StrOutputParser：AIMessage → str（不够）

• JsonOutputParser：AIMessage → Dict(JSON)

4.2 完整代码

from langchain_core.output_parsers import StrOutputParser from langchain_core.output_parsers import JsonOutputParser from langchain_core.prompts import PromptTemplate from langchain_community.chat_models.tongyi import ChatTongyi 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: str = chain.invoke({"lastname": "张", "gender": "女儿"}) print(res) print(type(res))

4.3 输入输出必须对齐

标准链类型流：

字典 → first_prompt → PromptValue → model → AIMessage → json_parser → 字典 → second_prompt → PromptValue → model → AIMessage → str_parser → 字符串

五、RunnableLambda：自定义函数入链

5.1 语法

除了固定功能的解析器，也可以自己编写 Lambda 完成自定义逻辑。RunnableLambda把普通函数转换为 Runnable 实例，方便自定义函数加入 chain。

语法：RunnableLambda(函数对象或 lambda 匿名函数)

from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableLambda from langchain_core.prompts import PromptTemplate from langchain_community.chat_models.tongyi import ChatTongyi str_parser = StrOutputParser() my_func = RunnableLambda(lambda ai_msg: {"name": ai_msg.content}) model = ChatTongyi(model="qwen3-max") first_prompt = PromptTemplate.from_template( "我邻居姓：{lastname}，刚生了{gender}，请起名，仅告知我名字，不要额外信息" ) second_prompt = PromptTemplate.from_template( "姓名{name}，请帮我解析含义。" ) chain = first_prompt | model | my_func | second_prompt | model | str_parser res: str = chain.invoke({"lastname": "张", "gender": "女儿"}) print(res) print(type(res))

5.2 函数也可以直接入链

chain = first_prompt | model | (lambda ai_msg: {"name": ai_msg.content}) | second_prompt | model | str_parser

因为 Runnable 在实现__or__时支持 Callable；函数就是 Callable 实例，本质是将函数自动转换为 RunnableLambda。

小结：

1. 将函数封装入RunnableLambda，它是 Runnable 接口实例，可以直接入链

2. 直接将函数入链，函数会自动转换为 RunnableLambda 对象

六、临时记忆：InMemoryChatMessageHistory

如果想要封装历史记录，除了自行维护历史消息外，也可以借助 LangChain 内置的历史记录功能，帮助模型在有历史记忆的情况下回答。

6.1 两个关键类

• 基于RunnableWithMessageHistory在原有链的基础上创建带有历史记录功能的新链（新 Runnable 实例）

• 基于InMemoryChatMessageHistory为历史记录提供内存存储（临时用）

from langchain_core.runnables.history import RunnableWithMessageHistory conversation_chain = RunnableWithMessageHistory( some_chain, # 被附加历史消息的 Runnable，通常是 chain None, # 获取指定会话 ID 的历史会话的函数 input_messages_key="input", # 用户输入在模板中的占位符 history_messages_key="chat_history" # 历史消息在模板中的占位符 ) chat_history_store = {} # 存放多个会话 ID 所对应的历史会话记录 def get_history(session_id): if session_id not in chat_history_store: chat_history_store[session_id] = InMemoryChatMessageHistory() return chat_history_store[session_id]

6.2 完整代码

from langchain_community.chat_models.tongyi import ChatTongyi from langchain_core.chat_history import InMemoryChatMessageHistory from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import PromptTemplate from langchain_core.runnables.history import RunnableWithMessageHistory def print_prompt(full_prompt): print("=" * 20, full_prompt.to_string(), "=" * 20) return full_prompt model = ChatTongyi(model="qwen3-max") prompt = PromptTemplate.from_template( "你需要根据对话历史回应用户问题。对话历史：{chat_history}。用户当前输入：{input}， 请给出回应" ) base_chain = prompt | print_prompt | model | StrOutputParser() chat_history_store = {} def get_history(session_id): if session_id not in chat_history_store: chat_history_store[session_id] = InMemoryChatMessageHistory() return chat_history_store[session_id] conversation_chain = RunnableWithMessageHistory( base_chain, get_history, input_messages_key="input", history_messages_key="chat_history" ) if __name__ == '__main__': session_config = {"configurable": {"session_id": "user_001"}} print(conversation_chain.invoke({"input": "小明有一只猫"}, session_config)) print(conversation_chain.invoke({"input": "小刚有两只狗"}, session_config)) print(conversation_chain.invoke({"input": "共有几只宠物？"}, session_config))

注意：若想在执行链的同时把提示词 print 出来，可在链中加入自定义函数；函数的输入应原封不动返回出去，避免破坏原有业务，仅在 return 之前 print 所需信息即可。

6.3 小结

• RunnableWithMessageHistory用于创建一个带有历史记忆功能的 Runnable 实例（链）

• 创建时需要提供BaseChatMessageHistory的具体实现

• InMemoryChatMessageHistory实现在内存中存储历史

七、长期记忆：自实现 FileChatMessageHistory

7.1 为什么内存不够

InMemoryChatMessageHistory仅在内存中临时存储，程序退出则记忆丢失。它继承自BaseChatMessageHistory。官方注释给出了实现指南，并给出基于文件的历史消息存储示例。可以自行实现基于 JSON 和本地文件的会话数据保存。

7.2 核心思路

• 基于文件存储会话记录，以session_id为文件名，不同 session 不同文件

• 继承BaseChatMessageHistory，实现 3 个方法：

  • add_messages：同步添加消息

  • messages：同步获取消息

  • clear：同步清除消息

import json, os from langchain_core.messages import messages_from_dict, message_to_dict # message_to_dict：单个消息对象（BaseMessage类实例） -> 字典 # messages_from_dict：[字典、字典...] -> [消息、消息...] # AIMessage、HumanMessage、SystemMessage 都是BaseMessage的子类 class FileChatMessageHistory(BaseChatMessageHistory): def __init__(self, session_id, storage_path): self.session_id = session_id # 会话id self.storage_path = storage_path # 不同会话id的存储文件，所在的文件夹路径 # 完整的文件路径 self.file_path = os.path.join(self.storage_path, self.session_id) # 确保文件夹是存在的 os.makedirs(os.path.dirname(self.file_path), exist_ok=True) def add_messages(self, messages: Sequence[BaseMessage]) -> None: # Sequence序列 类似list、tuple all_messages = list(self.messages) # 已有的消息列表 all_messages.extend(messages) # 新的和已有的融合成一个list # 将数据同步写入到本地文件中 # 类对象写入文件 -> 一堆二进制 # 为了方便，可以将BaseMessage消息转为字典（借助json模块以json字符串写入文件） # 官方message_to_dict：单个消息对象（BaseMessage类实例） -> 字典 # new_messages = [] # for message in all_messages: # d = message_to_dict(message) # new_messages.append(d) new_messages = [message_to_dict(message) for message in all_messages] # 将数据写入文件 with open(self.file_path, "w", encoding="utf-8") as f: json.dump(new_messages, f) @property # @property装饰器将messages方法变成成员属性用 def messages(self) -> list[BaseMessage]: # 当前文件内： list[字典] try: with open(self.file_path, "r", encoding="utf-8") as f: messages_data = json.load(f) # 返回值就是：list[字典] return messages_from_dict(messages_data) except FileNotFoundError: return [] def clear(self) -> None: with open(self.file_path, "w", encoding="utf-8") as f: json.dump([], f)

7.3 业务链部分

from langchain_core.prompts import PromptTemplate from langchain_core.runnables.history import RunnableWithMessageHistory from langchain_core.chat_history import BaseChatMessageHistory, BaseMessage from langchain_core.output_parsers import StrOutputParser from langchain_core.messages import messages_from_dict, message_to_dict from langchain_community.chat_models.tongyi import ChatTongyi from typing import Sequence, List import json llm = ChatTongyi(model="qwen3-max") prompt = PromptTemplate.from_template("""你是一个贴心的助手，需要根据对话历史回应用户的问题。 对话历史：{chat_history} 用户当前输入：{input} 你的回应：""") base_chain = prompt | llm | StrOutputParser() def get_message_history(session_id: str) -> BaseChatMessageHistory: """根据会话 ID 获取对应的对话历史存储实例""" return FileChatMessageHistory(session_id=session_id, storage_path="./chat_history") conversation_chain = RunnableWithMessageHistory( runnable=base_chain, get_session_history=get_message_history, input_messages_key="input", history_messages_key="chat_history", ) if __name__ == "__main__": session_config = {"configurable": {"session_id": "user_001"}} response1 = conversation_chain.invoke({"input": "小明有1只猫"}, config=session_config) print("第一轮：", response1) response2 = conversation_chain.invoke({"input": "小刚有2只狗"}, config=session_config) print("\n第二轮：", response2) response3 = conversation_chain.invoke( {"input": "小明和小刚一共有几只宠物?"}, config=session_config ) print("\n第三轮：", response3) # 测试程序重启后读取历史（注释上面的代码，单独运行下面的代码仍能获取历史） # response4 = conversation_chain.invoke( # {"input": "分别是什么宠物？"}, # config=session_config # ) # print("\n重启后第四轮：", response4)