Files
ai-agent-dev/08_链式组合.ipynb
2026-07-08 10:09:42 +08:00

559 lines
19 KiB
Plaintext
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# 08 链式组合\n",
"\n",
"## 学习目标\n",
"1. 深入理解 LCELLangChain Expression Language管道符 `|` 的组合原理\n",
"2. 掌握 `Runnable` 接口的核心方法:`invoke`、`batch`、`stream`、`ainvoke`\n",
"3. 学会在链中使用 `RunnableLambda` 进行数据格式转换\n",
"4. 理解 `RunnableParallel` 并行执行多个子链\n",
"5. 能够构建复杂的多步骤处理链"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 1. 什么是 LCEL 链式组合\n",
"\n",
"LCEL 是 LangChain 推荐的新一代链式组合语法。它用管道符 `|` 把各个组件串联起来,就像 Linux 命令行一样直观。\n",
"\n",
"```\n",
"chain = component_a | component_b | component_c\n",
"```\n",
"\n",
"数据流:\n",
"```\n",
"输入 -> component_a -> 中间结果 -> component_b -> 中间结果 -> component_c -> 输出\n",
"```\n",
"\n",
"LCEL 中的每个组件都实现了 `Runnable` 接口,它们都有统一的方法签名。"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 2. Runnable 接口的核心方法\n",
"\n",
"所有 LCEL 组件都实现了以下方法:\n",
"\n",
"| 方法 | 作用 |\n",
"| --- | --- |\n",
"| `invoke(input)` | 单条同步调用 |\n",
"| `batch(inputs)` | 批量同步调用 |\n",
"| `stream(input)` | 流式输出 |\n",
"| `ainvoke(input)` | 异步单条调用 |\n",
"| `abatch(inputs)` | 异步批量调用 |\n",
"\n",
"这些方法不仅单个组件有,组合后的链也有。"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from langchain_openai import ChatOpenAI\n",
"from langchain_core.prompts import ChatPromptTemplate\n",
"from langchain_core.output_parsers import StrOutputParser\n",
"from dotenv import load_dotenv\n",
"\n",
"load_dotenv()\n",
"\n",
"\n",
"llm = ChatOpenAI(model='qwen3.6-35b-A3b', temperature=0.7)\n",
"\n",
"# 构建一个简单链\n",
"prompt = ChatPromptTemplate.from_messages([\n",
" ('system', '你是一个翻译助手。'),\n",
" ('user', '把以下中文翻译成英文:{text}')\n",
"])\n",
"\n",
"chain = prompt | llm | StrOutputParser()\n",
"\n",
"# 使用 invoke 单条调用\n",
"result = chain.invoke({'text': '人工智能'})\n",
"print('invoke 结果:', result)\n",
"\n",
"# 使用 batch 批量调用\n",
"results = chain.batch([\n",
" {'text': '苹果'},\n",
" {'text': '香蕉'},\n",
" {'text': '机器学习'}\n",
"])\n",
"print('\\nbatch 结果:', results)\n",
"\n",
"# 使用 stream 流式输出\n",
"print('\\nstream 结果:')\n",
"for chunk in chain.stream({'text': '未来科技'}):\n",
" print(chunk, end='')\n",
"print()"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### 代码解释\n",
"\n",
"- `prompt` 接收字典,输出格式化后的消息列表\n",
"- `llm` 接收消息列表,输出 `AIMessage`\n",
"- `StrOutputParser()` 接收 `AIMessage`,输出字符串\n",
"- 组合后的 `chain` 继承了三个组件的能力,可以直接调用 `invoke`、`batch`、`stream`"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 3. 使用 RunnableLambda 转换数据\n",
"\n",
"在组合链时,前一个组件的输出格式可能和后一个组件的输入格式不匹配。\n",
"\n",
"例如:前一个链输出字符串,后一个链需要字典。这时可以用 `RunnableLambda` 或普通 lambda 函数做转换。"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from langchain_openai import ChatOpenAI\n",
"from langchain_core.prompts import ChatPromptTemplate\n",
"from langchain_core.output_parsers import StrOutputParser\n",
"from langchain_core.runnables import RunnableLambda\n",
"\n",
"llm = ChatOpenAI(model='qwen3.6-35b-A3b', temperature=0.7)\n",
"\n",
"# 第一步:把中文翻译成英文\n",
"translate_prompt = ChatPromptTemplate.from_messages([\n",
" ('system', '你是一个翻译助手。'),\n",
" ('user', '把以下中文翻译成英文:{text}')\n",
"])\n",
"translate_chain = translate_prompt | llm | StrOutputParser()\n",
"\n",
"# 第二步:把英文总结成一句话\n",
"summarize_prompt = ChatPromptTemplate.from_messages([\n",
" ('system', '你是一个总结助手。'),\n",
" ('user', '请用一句话总结以下内容:{text}')\n",
"])\n",
"summarize_chain = summarize_prompt | llm | StrOutputParser()\n",
"\n",
"# 翻译链输出字符串,但总结链需要字典 {text: ...}\n",
"# 使用 RunnableLambda 做格式转换\n",
"def to_dict(text):\n",
" return {'text': text}\n",
"\n",
"# 组合链:翻译 -> 转字典 -> 总结\n",
"full_chain = translate_chain | RunnableLambda(to_dict) | summarize_chain\n",
"\n",
"result = full_chain.invoke({'text': '大语言模型开发效率比较高,而且比较聪明,而且成本比较低,大家都在学习怎么使用,正在改变软件的开发方式。'})\n",
"print('最终结果:', result)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### 代码解释\n",
"\n",
"| 组件 | 输入 | 输出 |\n",
"| --- | --- | --- |\n",
"| `translate_chain` | `{'text': '...'}` | 英文字符串 |\n",
"| `RunnableLambda(to_dict)` | 英文字符串 | `{'text': '...'}` |\n",
"| `summarize_chain` | `{'text': '...'}` | 总结字符串 |\n",
"\n",
"`RunnableLambda` 把一个普通函数包装成 Runnable可以无缝接入链中。"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 4. 直接传递多个变量RunnableParallel\n",
"\n",
"`RunnableParallel` 允许你同时运行多个子链,并把多个结果合并成一个字典。\n",
"\n",
"它适合的场景:\n",
"- 同时让模型做翻译和总结\n",
"- 同时提取不同类型的信息\n",
"- 一个输入需要生成多个输出"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from langchain_openai import ChatOpenAI\n",
"from langchain_core.prompts import ChatPromptTemplate\n",
"from langchain_core.output_parsers import StrOutputParser\n",
"from langchain_core.runnables import RunnableParallel\n",
"\n",
"llm = ChatOpenAI(model='qwen3.6-35b-A3b', temperature=0.7)\n",
"\n",
"# 定义多个子链\n",
"translate_prompt = ChatPromptTemplate.from_messages([\n",
" ('system', '你是翻译助手。'),\n",
" ('user', '把以下中文翻译成英文:{text}')\n",
"])\n",
"translate_chain = translate_prompt | llm | StrOutputParser()\n",
"\n",
"summary_prompt = ChatPromptTemplate.from_messages([\n",
" ('system', '你是总结助手。'),\n",
" ('user', '用一句话总结:{text}')\n",
"])\n",
"summary_chain = summary_prompt | llm | StrOutputParser()\n",
"\n",
"keywords_prompt = ChatPromptTemplate.from_messages([\n",
" ('system', '你是关键词提取助手。'),\n",
" ('user', '提取3个关键词用逗号分隔{text}')\n",
"])\n",
"keywords_chain = keywords_prompt | llm | StrOutputParser()\n",
"\n",
"# 并行执行三个子链,输入都是 {'text': '...'}\n",
"parallel_chain = RunnableParallel(\n",
" translation=translate_chain,\n",
" summary=summary_chain,\n",
" keywords=keywords_chain\n",
")\n",
"\n",
"result = parallel_chain.invoke({'text': '人工智能正在改变教育、医疗和交通行业'})\n",
"\n",
"print('返回类型:', type(result))\n",
"print('翻译结果:', result['translation'])\n",
"print('总结结果:', result['summary'])\n",
"print('关键词结果:', result['keywords'])"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### 代码解释\n",
"\n",
"- `RunnableParallel(...)` 接收一个字典,键是输出字段名,值是子链\n",
"- 所有子链会同时接收同一个输入 `{'text': '...'}`\n",
"- 返回结果是一个字典,包含每个子链的输出\n",
"- 适合需要「一次输入、多种处理」的场景"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 5. 组合并行与串行\n",
"\n",
"`RunnableParallel` 的输出是一个字典,可以继续传给下一个组件使用。\n",
"\n",
"例如:先并行做翻译和总结,再把结果合并成一个报告。"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from langchain_openai import ChatOpenAI\n",
"from langchain_core.prompts import ChatPromptTemplate\n",
"from langchain_core.output_parsers import StrOutputParser\n",
"from langchain_core.runnables import RunnableParallel\n",
"\n",
"llm = ChatOpenAI(model='qwen3.6-35b-A3b', temperature=0.7)\n",
"\n",
"# 并行生成翻译和总结\n",
"parallel_chain = RunnableParallel(\n",
" translation=(\n",
" ChatPromptTemplate.from_messages([\n",
" ('system', '翻译助手'),\n",
" ('user', '把中文翻译成英文:{text}')\n",
" ]) | llm | StrOutputParser()\n",
" ),\n",
" summary=(\n",
" ChatPromptTemplate.from_messages([\n",
" ('system', '总结助手'),\n",
" ('user', '用一句话总结:{text}')\n",
" ]) | llm | StrOutputParser()\n",
" )\n",
")\n",
"\n",
"# 把并行结果合并成最终报告\n",
"report_prompt = ChatPromptTemplate.from_messages([\n",
" ('system', '你是一个报告生成助手。'),\n",
" ('user', '''请根据以下翻译和总结生成一份简短报告:\\n\\n翻译{translation}\\n总结{summary}\\n\\n请输出包含「原文概要」和「英文翻译」两部分的报告。''')\n",
"])\n",
"\n",
"full_chain = parallel_chain | report_prompt | llm | StrOutputParser()\n",
"\n",
"result = full_chain.invoke({'text': '大语言模型开发效率比较高,而且比较聪明,而且成本比较低,大家都在学习怎么使用,正在改变软件的开发方式。'})\n",
"print(result)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### 代码解释\n",
"\n",
"| 阶段 | 说明 |\n",
"| --- | --- |\n",
"| `parallel_chain` | 同时生成翻译和总结,输出 `{'translation': ..., 'summary': ...}` |\n",
"| `report_prompt` | 接收字典,用 `{translation}` 和 `{summary}` 填充模板 |\n",
"| `llm \\| StrOutputParser()` | 生成最终报告 |\n",
"\n",
"这种「先并行、再串行」的模式在实际应用中非常常见。"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 6. 条件分支RunnableBranch\n",
"\n",
"`RunnableBranch` 可以根据条件选择不同的子链执行。\n",
"\n",
"例如:根据输入语言类型选择不同的处理链。"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from langchain_openai import ChatOpenAI\n",
"from langchain_core.prompts import ChatPromptTemplate\n",
"from langchain_core.output_parsers import StrOutputParser\n",
"from langchain_core.runnables import RunnableBranch, RunnableLambda\n",
"\n",
"llm = ChatOpenAI(model='qwen3.6-35b-A3b', temperature=0.1)\n",
"\n",
"# 定义两个处理链\n",
"chinese_chain = (\n",
" ChatPromptTemplate.from_messages([\n",
" ('system', '你是中文助手。'),\n",
" ('user', '请用中文解释:{text}')\n",
" ]) | llm | StrOutputParser()\n",
")\n",
"\n",
"english_chain = (\n",
" ChatPromptTemplate.from_messages([\n",
" ('system', 'You are an English assistant.'),\n",
" ('user', 'Please explain in English: {text}')\n",
" ]) | llm | StrOutputParser()\n",
")\n",
"\n",
"# 判断输入文本是否包含中文字符\n",
"def is_chinese(input_dict):\n",
" text = input_dict['text']\n",
" return any('\\u4e00' <= char <= '\\u9fff' for char in text)\n",
"\n",
"# 构建分支链\n",
"branch_chain = RunnableBranch(\n",
" (is_chinese, chinese_chain),\n",
" english_chain # 默认分支\n",
")\n",
"\n",
"print('中文输入结果:')\n",
"print(branch_chain.invoke({'text': '神经网络'})[:250])\n",
"\n",
"print('\\n英文输入结果')\n",
"print(branch_chain.invoke({'text': 'neural network'})[:250])\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### 代码解释\n",
"\n",
"- `RunnableBranch` 接收多个 `(条件函数, 子链)` 元组,最后是默认子链\n",
"- 条件函数接收链的输入,返回 `True` 或 `False`\n",
"- 从上往下匹配,第一个条件为 `True` 的分支会被执行\n",
"- 如果没有匹配,执行默认分支\n",
"\n",
"注意:条件函数处理的输入是整个链的输入(这里是字典),不是前一个组件的输出。"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 7. 查看链的结构\n",
"\n",
"复杂的链可以通过 `get_graph().print_ascii()` 方法查看结构。它会用 ASCII 字符打印出链中的节点和连接关系,非常直观。\n",
"\n",
"> 需要安装依赖:`pip install grandalf`\n",
"> 安装后请**重启 Jupyter 内核**再运行本单元格。"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from langchain_openai import ChatOpenAI\n",
"from langchain_core.prompts import ChatPromptTemplate\n",
"from langchain_core.output_parsers import StrOutputParser\n",
"\n",
"llm = ChatOpenAI(model='qwen3.6-35b-A3b', temperature=0.7)\n",
"\n",
"chain = (\n",
" ChatPromptTemplate.from_messages([\n",
" ('user', '把以下内容翻译成英文:{text}')\n",
" ])\n",
" | llm\n",
" | StrOutputParser()\n",
" | (lambda x: {'text': x})\n",
" | ChatPromptTemplate.from_messages([\n",
" ('user', '用一句话总结:{text}')\n",
" ])\n",
" | llm\n",
" | StrOutputParser()\n",
")\n",
"\n",
"# 打印链的 ASCII 结构图\n",
"chain.get_graph().print_ascii()"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### 代码解释\n",
"\n",
"- `get_graph()` 获取链的图结构对象\n",
"- `print_ascii()` 用 ASCII 字符画出节点和边\n",
"- 可以清楚地看到数据从输入到输出经过了哪些步骤\n",
"- 如果提示缺少 `grandalf`,请先安装:`pip install grandalf`,然后重启 Jupyter 内核"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 8. 完整示例:智能客服工单处理\n",
"\n",
"下面是一个综合示例:用户输入一条投诉或咨询,系统同时完成情绪分析、分类、生成回复建议。"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from langchain_openai import ChatOpenAI\n",
"from langchain_core.prompts import ChatPromptTemplate\n",
"from langchain_core.output_parsers import StrOutputParser\n",
"from langchain_core.runnables import RunnableParallel\n",
"\n",
"llm = ChatOpenAI(model='qwen3.6-35b-A3b', temperature=0.3)\n",
"\n",
"# 并行分析:情绪 + 分类 + 摘要\n",
"# 注意reply_prompt 中还需要原始输入 text所以这里用 lambda 把 text 一起传下去\n",
"analysis_chain = RunnableParallel(\n",
" text=lambda x: x['text'],\n",
" sentiment=(\n",
" ChatPromptTemplate.from_messages([\n",
" ('system', '情绪分析助手,只输出:正面/负面/中性'),\n",
" ('user', '{text}')\n",
" ]) | llm | StrOutputParser()\n",
" ),\n",
" category=(\n",
" ChatPromptTemplate.from_messages([\n",
" ('system', '分类助手,只输出类别:售后/产品/物流/其他'),\n",
" ('user', '{text}')\n",
" ]) | llm | StrOutputParser()\n",
" ),\n",
" summary=(\n",
" ChatPromptTemplate.from_messages([\n",
" ('system', '摘要助手,用一句话概括'),\n",
" ('user', '{text}')\n",
" ]) | llm | StrOutputParser()\n",
" )\n",
")\n",
"\n",
"# 根据分析结果生成回复建议\n",
"reply_prompt = ChatPromptTemplate.from_messages([\n",
" ('system', '你是客服主管,请根据分析结果生成回复建议。'),\n",
" ('user', '''用户反馈:{text}\n",
"\n",
"分析结果:\n",
"- 情绪:{sentiment}\n",
"- 类别:{category}\n",
"- 摘要:{summary}\n",
"\n",
"请生成一段礼貌、专业的回复建议。''')\n",
"])\n",
"\n",
"full_chain = analysis_chain | reply_prompt | llm | StrOutputParser()\n",
"\n",
"result = full_chain.invoke({\n",
" 'text': '你们的发货速度太慢了,说好三天到,结果等了一周,非常失望!'\n",
"})\n",
"print(result)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### 代码解释\n",
"\n",
"| 步骤 | 说明 |\n",
"| --- | --- |\n",
"| `analysis_chain` | 并行执行三个分析任务,输出字典 |\n",
"| `reply_prompt` | 把原始输入和分析结果一起填充到回复模板 |\n",
"| `llm \\| StrOutputParser()` | 生成最终回复建议 |\n",
"\n",
"注意:`reply_prompt` 中既用到了原始输入 `{text}`,也用到了并行分析结果 `{sentiment}`、`{category}`、`{summary}`。"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 9. 本节课练习\n",
"\n",
"1. 用 `|` 构建一个链:输入中文 -> 翻译为英文 -> 再翻译回中文,观察「回译」后的变化\n",
"2. 使用 `RunnableLambda` 把字符串输出转换为字典,再接第二个 Prompt 模板\n",
"3. 使用 `RunnableParallel` 让一个输入同时生成「正式版」和「口语版」两种翻译\n",
"4. 使用 `RunnableBranch` 根据输入长度选择不同模型:短文本用轻量提示,长文本用详细提示\n",
"5. 用 `chain.get_graph().print_ascii()` 查看你自己构建的链的结构\n"
]
}
],
"metadata": {
"kernelspec": {
"display_name": ".venv",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.11.3"
}
},
"nbformat": 4,
"nbformat_minor": 4
}