AI 工程化实战：从零手搓代码，这一次彻底搞懂MCP！撇
简介
langchain中提供的chain链组件，能够帮助我门快速的实现各个组件的流水线式的调用，和模型的问答
Chain链的组成
根据查阅的资料，langchain的chain链结构如下：
$$Input \rightarrow Prompt \rightarrow Model \rightarrow Output$$
其中langchain框架提供了几个常用构建chain链的工具：
工具名称 作用 流程
RunnablePassthrough 传递原本的数据或添加新的字段 $$A->B$$
RunnableParallel 并发输出结果并将结果同时传递 $$A,B->C$$
RunnableLambda 自定义传递工具
乍一看很疑惑，我接下来用案例来解释各种用法。
构建较为复杂的chain链
这个案例几乎用了上面所有工具，用于演示用法
案例
案例描述：输入论文的话题，写一篇950字的高中论文。
import os
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableParallel
#你的qwen模型apikey
os.environ["DASHSCOPE_API_KEY"] = "apikey"
model = ChatTongyi(model="qwen-max")
outline_prompt = ChatPromptTemplate.from_template(
"请给主题为 {topic} 的议论文写一个 总-递进-总 的简短大纲，一共分为5段。"
)
outline_chain = outline_prompt | model | StrOutputParser()
def mock_search(input_data):
return """
1. 利：Google Health AI 筛查乳腺癌准确率超人类。
2. 利：AlphaFold 预测蛋白质结构，缩短科研周期。
3. 弊：GPT-4 普及导致初级文案、原画设计岗位萎缩。
4. 弊：Deepfake 技术被用于电信诈骗和虚假视频。
"""
output_prompt = ChatPromptTemplate.from_template(
"你是一位高考作文专家。请基于大纲：\n{outline}\n并结合以下案例素材：\n{data}\n"
"就主题【{topic}】写一篇高考论文。要求：950字左右，论证严密，文采斐然。"
)
output_chain = output_prompt | model | StrOutputParser()
complex_chain = (
RunnableParallel({
"outline": outline_chain,
"data": mock_search,
"topic": RunnablePassthrough()
})
| output_chain
)
topic_input = "AI 进步的利与弊：在智能时代保持人类的温度"
print(f"正在为您撰写关于《{topic_input}》的论文...\n")
final_essay = complex_chain.invoke({"topic": topic_input})
print(final_essay)
代码解释
其他的代码我上期解释了，这里就不废话了，我着重讲chain链的构建，总体chain链的流程如下：
输入话题->获取写作的大纲 ──╮
├─? 根据大纲和示例写一篇论文
查询相关的示例 ──╯
根据输入流程图流程，我们个以分解成一个个相关的链，再将各个链串起来。
构建各部分的链
1.获取写作大纲
outline_prompt = ChatPromptTemplate.from_template(
"请给主题为 {topic} 的议论文写一个 总-递进-总 的简短大纲，一共分为5段。"
)
outline_chain = outline_prompt | model | StrOutputParser()
流程描述：构建prompt->喂给ai->将返回解析成文本
这里用的ChatPromptTemplate.from_template和上的ChatPromptTemplate.from_messages不同，区别在于前者比较简单，相当于后者直接用user字典的形式，后者from_messages有langchain框架提供的prompt模板
2.查询相关的示例
这里就直接用Gemini，mock一些模拟数据（不保证真），用于完成案例，实际情况可以自己完善搜索逻辑。
def mock_search(input_data):
return """
1. 利：Google Health AI 筛查乳腺癌准确率超人类。
2. 利：AlphaFold 预测蛋白质结构，缩短科研周期。
3. 弊：GPT-4 普及导致初级文案、原画设计岗位萎缩。
4. 弊：Deepfake 技术被用于电信诈骗和虚假视频。
"""
这个相当于RanableLamda,后面可以不用显示调用RanableLamda(mock_search)
3. 根据大纲和示例写一篇论文
output_prompt = ChatPromptTemplate.from_template(
"你是一位高考作文专家。请基于大纲：\n{outline}\n并结合以下案例素材：\n{data}\n"
"就主题【{topic}】写一篇高考论文。要求：950字左右，论证严密，文采斐然。"
)
output_chain = output_prompt | model | StrOutputParser()
流程描述：构建prompt->喂给ai->将返回解析成文本
将各个链连起来
complex_chain = (
RunnableParallel({
"outline": outline_chain,
"data": mock_search,
"topic": RunnablePassthrough()
})
| output_chain
)
这里利用RunnableParallel将获取写作的大纲和查询相关的示例两个流程并列运行后一起输出到后面，传递给output_chain继续处理。
问题
我不想利用RunnableParallel行不行？ 当然可以，可以用线性来代替，先查资料，后写大纲，然后再进行文章输出，但是效率可能会比较慢。
我希望看到输出的data和outline字段怎么办？ ，可以利用RunnablePassthrough().assign将生成的文本保存在新的字段中，调用时根据字典的方式定位各个文本，如下：
complex_chain = (
RunnableParallel({
"outline": outline_chain,
"data": mock_search,
"topic": RunnablePassthrough()
})
| RunnablePassthrough().assign(essay=output_chain)
)
print(response['essay'])
print(response['data'])
......幽谖购诹