{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 08 链式组合\n",
    "\n",
    "## 学习目标\n",
    "1. 深入理解 LCEL（LangChain 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
}