第 7 章 LangChain 核心概念：LCEL、Runnable 与可组合管道

如果你写过 Java Stream 的 map/filter/toList，或者用 Function.compose 做过函数组合，那么本章的主角——LangChain 的可组合管道（LCEL）——对你来说会非常眼熟。LangChain 不是"AI 魔法"，它的核心其实是一套接口约定和管道组装规则。理解这套规则，你就能用工程思维驾驭它，而不是被它牵着走。

版本说明：本教程统一采用 langchain 0.3.x（langchain、langchain-openai、langchain-community 均为 0.3，pydantic 为 2.x）。LangChain 1.0 已于 2025 年正式发布，API 有进一步收敛与调整；本书所有代码基于 0.3 编写，涉及新版本迁移时一律以官方文档为准。

本章你将学到

• LangChain 的定位：它是什么、不是什么，以及社区争议的真相。
• 六个核心抽象：ChatOpenAI、ChatPromptTemplate、OutputParser、Retriever、Memory、Tool。
• LCEL 管道的组合符 | 与 Runnable 一族（Passthrough / Lambda / Parallel）。
• 三种执行方式：invoke / stream / batch。
• 一个完整可运行的翻译链，以及一个"拆分子任务再合并"的并行示例。
• 与第 5 章手写实现（参见第 5 章）的客观对比。
• 记忆（Memory）在 0.3 中的推荐做法：显式管理 messages。
• 流式与异步：.stream() 与 .astream()。

本章是全书的分水岭：第 5 章我们徒手写过一个最小 Agent，第 8 章的工具调用与第 9-10 章的 LangGraph，全部建立在本章的 Runnable 与 LCEL 之上。

7.1 LangChain 是什么

7.1.1 定位：可组合的工具箱，不是万能框架

一句话：LangChain 是一套围绕 LLM 应用的"可组合构建块 + 集成生态"，而不是一个替你思考的框架。它解决的核心工程问题是：LLM 应用里充满了重复的"胶水代码"——拼提示词、调用模型、解析输出、接向量库、接工具、维护多轮历史——LangChain 把每一段胶水都抽象成一个实现了 Runnable 协议的组件，然后用管道组合起来。

7.1.2 历史与版本演进（一句话）

LangChain 于 2022 年底诞生，走了这样一条路：0.1（2024 年初） 确立核心 API 与 LCEL 管道；0.2（2024 年中） 把 langchain 拆分为 langchain-core / langchain / langchain-community，各模型厂商包独立发布；0.3（2024 年底） 合并 core、清理大量 0.1 时代的高层遗留 API，是当前最稳定、最值得学习的基线；1.0（2025 年 GA） 进一步收敛 API 并官方化 create_* 工厂函数模式。本书以 0.3 为准。

7.1.3 为什么有人吐槽（客观看待）

业界对 LangChain 的批评是真的，你应该知道它们，从而在选型时做出判断：

1. 抽象过多、层级混乱：0.1 时代有 LLMChain、SequentialChain、AgentExecutor 等一大堆高层类，"一行代码做 Agent"沦为黑盒。0.3 已把这些标记为 deprecated，官方立场是"一律用 LCEL 重写"，但存量代码里仍然遍地都是。
2. 黑盒难调试：高层链包装了太多细节，出了问题你常常不知道发给模型的请求体到底长什么样——这对工程师来说是灾难。0.3 提供了 get_openai_callback、回调（callbacks）和 LangSmith 链路追踪，但需要主动开启。
3. 版本动荡：0.1 → 0.2 → 0.3 → 1.0 多次破坏性变更。典型案例：ChatOpenAI 从 langchain-community 挪到 langchain-openai；RunnableWithMessageHistory 的参数反复调整；1.0 又删掉一批旧 API。追新版本的成本很高。
4. 依赖体积：装一个 langchain 会拖进大量传递依赖，对小项目显得笨重。

客观地讲：吐槽主要针对高层遗留抽象；而 LCEL 这个底层设计（Runnable 协议 + 管道组合）反而是被广泛认可的——它足够简单，甚至脱离 LangChain 也能独立使用（langchain-core 可以单独安装）。

7.1.4 什么时候值得用，什么时候别用

• 值得用：需要接多个生态（向量库、工具、文档加载器、不同模型厂商）；要统一处理流式/异步/批量的执行协议；要快速搭 Agent + RAG 的原型并保持可演进。
• 别用：只有"单次调用 + 简单提示词"；需要精确控制发给模型的每个字节（wire format）；对依赖体积敏感、想极致轻量；纯粹想学习底层原理（此时手写，参见第 5 章）。

核心立场：小系统可以手写，大系统用框架，但无论用哪种，都必须理解底层协议。

7.2 六个核心抽象

先看环境准备。三个包缺一不可，langchain-core 会被自动带入：

pip install langchain==0.3.* langchain-openai==0.3.* langchain-community==0.3.* pydantic==2.*

所有示例统一使用如下模型封装，天然兼容国产 OpenAI 兼容端点（DeepSeek / Qwen / GLM）：

import os

from langchain_openai import ChatOpenAI

# 模型封装。只需要改环境变量即可切换厂商：
#   DeepSeek:  base_url=https://api.deepseek.com/v1        model=deepseek-chat
#   通义千问:  base_url=https://dashscope.aliyuncs.com/compatible-mode/v1  model=qwen-plus
#   GLM:       base_url=https://open.bigmodel.cn/api/paas/v4             model=glm-4-plus
model = ChatOpenAI(
    model=os.getenv("LLM_MODEL", "deepseek-chat"),
    api_key=os.getenv("OPENAI_API_KEY"),  # 必须设置 OPENAI_API_KEY
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.deepseek.com/v1"),
    temperature=0.2,
)

Windows（cmd）下可这样设置环境变量后运行：

set OPENAI_API_KEY=sk-xxxxxxxx
set OPENAI_BASE_URL=https://api.deepseek.com/v1
set LLM_MODEL=deepseek-chat

macOS / Linux 把 set 换成 export 即可。

7.2.1 ChatOpenAI：模型封装

ChatOpenAI 本质上是一个"模型客户端 + Chat 消息协议"的封装：输入一组消息（role + content），输出一个 AIMessage。它等价于你在手写实现里维护的"HTTP 调用 + JSON 序列化 + 错误处理"那一坨代码。

from langchain_core.messages import HumanMessage, SystemMessage

resp = model.invoke([
    SystemMessage(content="你是一位 Java 架构师，回答要简洁。"),
    HumanMessage(content="用一句话解释什么是 Backpressure。"),
])
print(type(resp))          # <class 'langchain_core.messages.ai.AIMessage'>
print(resp.content)        # 模型回复的文本

要点：invoke 的入参是消息列表（或字符串），出参是 AIMessage 对象。后面你会看到，StrOutputParser 专门负责把 AIMessage 拆成字符串——这就是"转换器"存在的原因。

7.2.2 ChatPromptTemplate：模板引擎 + 参数绑定

ChatPromptTemplate 之于提示词，相当于 Thymeleaf 之于 HTML：模板 + 参数绑定。它管理 system / user / assistant 多种角色的消息结构，并在运行时注入变量。

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位资深中英翻译，译文准确且自然。"),
    ("user", "请把下面的中文翻译成英文：\n{text}"),
])

# 参数绑定：等价于把 {text} 替换为实际值
rendered = prompt.invoke({"text": "分布式系统需要关注一致性。"})
print(rendered.to_messages())  # 渲染后的完整消息列表

从 from_messages 传入的二元组（role, content）会自动转成对应消息对象；支持 {变量} 占位符，还支持可迭代的 MessagesPlaceholder（第 7.6 节会用到）。

7.2.3 OutputParser：把模型输出变成我们要的类型

模型输出是文本（或结构化消息），而业务代码往往需要 String / dict / Java 里的 POJO。OutputParser 就是那层转换：

• StrOutputParser：最常用，直接取 AIMessage.content 返回字符串。
• JsonOutputParser：解析模型输出的 JSON（支持流式部分解析）。
• CommaSeparatedListOutputParser：把逗号分隔文本解析成列表。
• PydanticOutputParser：把输出解析成 pydantic 对象（0.1 时代的方案）。

from langchain_core.output_parsers import StrOutputParser

# 转换器：AIMessage -> str
parser = StrOutputParser()
text = parser.invoke(resp)  # resp 是上面的 AIMessage
print(text)

注意：在 0.3 中，如果你的目标是"拿到强类型的结构化输出"，官方推荐不再用手写 JSON 解析，而是用 model.with_structured_output(你的 pydantic 模型)——它内部走工具调用（function calling），可靠得多（详见第 8 章）：

from pydantic import BaseModel, Field


class Translation(BaseModel):
    en: str = Field(description="英文译文")
    back: str = Field(description="中文回译")


structured_model = model.with_structured_output(Translation)
out = structured_model.invoke([("user", "翻译：高可用")])
print(out.en)  # 直接拿到字段，不用自己解析 JSON

7.2.4 Retriever：检索器（详见第 8 章）

Retriever 是 RAG 的查询抽象：输入一个查询文本，返回一组相关文档。它等价于"搜索/相似度检索"的 DAO 接口。这里只给出一个最小示意，落地细节（Embedding 选型、向量库、混合检索）在参见第 8 章展开：

from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

# 注意：Embedding 厂商与 Chat 模型不一定一致，第 8 章会讲国产端点下的选型
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts(
    ["今天上海多云，最高 26 度。", "今天北京晴，最高 30 度。"],
    embedding=embeddings,
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

docs = retriever.invoke("北京今天天气如何？")
print([d.page_content for d in docs])

7.2.5 Memory：记忆

Memory 负责把多轮对话历史带进下一次调用。但正如 7.6 节将要详述的，LangChain 0.3 明确不推荐再用旧的 Memory 类，而是建议显式管理 messages，因此这里先不展开。

7.2.6 Tool：工具

Tool 是向模型暴露"可调用能力"的声明式接口——你的 Agent 可以调用 get_weather、查询订单、调外部 API。等价于把你的 Java 服务方法注册成模型可感知的调用清单。详细机制（function calling、参数 schema 推导）在第 8 章，这里给出最简形态：

from langchain_core.tools import tool


@tool
def get_weather(city: str) -> str:
    """查询指定城市的实时天气。"""
    # 真实场景这里会调气象 API；此处返回假数据演示
    return f"{city}：晴，26℃"


print(get_weather.name)         # get_weather
print(get_weather.description)  # 查询指定城市的实时天气。
print(get_weather.args_schema)  # 由类型注解自动推导的 pydantic 参数模型

@tool 装饰器会读取函数名、docstring、参数类型注解，自动生成 OpenAI/DeepSeek/Qwen/GLM 都能识别的 function schema——这正是第 8 章 model.bind_tools([...]) 的输入。

7.3 LCEL 管道：LangChain 的骨架

LCEL 全称 LangChain Expression Language（LangChain 表达式语言）。它不是一门新语言，而是用 Python 的操作符重载实现的管道组装语法：A | B 的语义是"先执行 A，把 A 的输出作为 B 的输入"。

cat access.log | grep "ERROR" | wc -l        # Unix 管道
list.stream().map(x -> ...).filter(...)      # Java Stream
prompt | model | StrOutputParser()           # LCEL

7.3.1 组合符 | 的语义

A | B 在 Python 中实际调用的是 A.__or__(B)，返回一个 RunnableSequence（A 和 B 按顺序执行）。关键规则只有一条：

管道中每一段的输出形状，必须等于下一段的输入形状。（就像 Function<T,R> 组合时，前者的 R 必须是后者的 T。）

这也是 7.4 节所有示例中 StrOutputParser 的意义：model 输出 AIMessage，而 StrOutputParser 输出 str，把形状"翻译"成下游要的形态。

LCEL 的价值不仅在于写法简洁，更在于：任意实现了 Runnable 协议的对象，都拥有统一执行协议——invoke（单次）、batch（批量）、stream（流式）、以及各自的异步版本 ainvoke / abatch / astream。你在手写实现里为"支持流式"写的额外分支，在 LCEL 里是每个组件默认自带的。

7.3.2 Runnable 一族

RunnablePassthrough：透传

from langchain_core.runnables import RunnablePassthrough

# 把输入字符串原样填到 {text} 槽位，从而支持 chain.invoke("一句话")
chain = {"text": RunnablePassthrough()} | prompt | model | StrOutputParser()
print(chain.invoke("高并发系统需要关注缓存与限流。"))

# assign: 保留已有字段，追加计算字段（等价于给对象加派生属性）
chain2 = RunnablePassthrough.assign(text_len=lambda d: len(d["text"]))
print(chain2.invoke({"text": "hello"}))  # {'text': 'hello', 'text_len': 5}

RunnableLambda：包 Python 函数

from langchain_core.runnables import RunnableLambda


def normalize(data: dict) -> str:
    """清洗输入：折叠多余空白。等价于一个自定义的 pre-processor。"""
    return " ".join(data["text"].split())


normalize_chain = RunnableLambda(normalize)
print(normalize_chain.invoke({"text": "  高并发   系统  "}))  # "高并发 系统"

RunnableLambda 让你在管道任意位置插入普通业务函数（数据清洗、调用你现有的 Java/其他服务、拼装中间结果），无需重写管道——这往往是从手写迁移到 LangChain 后第一个被用起来的组件。

RunnableParallel：并行 / 分支

from langchain_core.runnables import RunnableParallel

# 两条分支共享同一输入 dict，各自独立处理
parallel = RunnableParallel(
    text_len=lambda d: len(d["text"]),
    text_upper=lambda d: d["text"].upper(),
)
out = parallel.invoke({"text": "langchain"})
print(out)  # {'text_len': 9, 'text_upper': 'LANGCHAIN'}

两条分支互不依赖、可并行执行，最终合并成一个 dict。这正是 7.4 节"拆分子任务再合并"的骨架。

7.3.3 三种执行方式：invoke / stream / batch

同一个链，天然支持三种执行方式：

# 1) invoke：单次调用，返回完整结果
r = chain.invoke({"text": "消息队列解决什么问题？"})

# 2) stream：流式返回，边生成边消费（对 WebSSE 场景是刚需）
for chunk in chain.stream({"text": "消息队列解决什么问题？"}):
    print(chunk, end="", flush=True)

# 3) batch：批量调用，内部可能合并请求
results = chain.batch([
    {"text": "第一条"},
    {"text": "第二条"},
])

三者共用同一个管道定义，无需为流式写特判——这就是 LCEL 统一执行协议的意义。

7.4 完整示例

7.4.1 示例一：翻译链（prompt | model | StrOutputParser）

完整可运行。把 OPENAI_API_KEY 指向任一 OpenAI 兼容端点即可（见 7.2 的环境变量说明）。

import os

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


def build_model() -> ChatOpenAI:
    """构建模型客户端，兼容 DeepSeek / Qwen / GLM 等国产端点。"""
    return ChatOpenAI(
        model=os.getenv("LLM_MODEL", "deepseek-chat"),
        api_key=os.getenv("OPENAI_API_KEY"),  # 必须设置，否则运行时抛错
        base_url=os.getenv("OPENAI_BASE_URL", "https://api.deepseek.com/v1"),
        temperature=0.2,
    )


def main():
    model = build_model()

    # 模板：等价于模板引擎 + 参数绑定
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一位资深技术翻译，精通中英互译，译文准确且自然。"),
        ("user", "请把下面的中文翻译成英文：\n{text}"),
    ])

    # LCEL 管道：把三个组件串起来（输出形状：dict -> AIMessage -> str）
    chain = prompt | model | StrOutputParser()

    # 单次调用
    print("=== invoke ===")
    print(chain.invoke({"text": "高并发系统需要关注缓存与限流。"}))

    # 流式调用
    print("\n=== stream ===")
    for chunk in chain.stream({"text": "Raft 协议如何保证一致性？"}):
        print(chunk, end="", flush=True)
    print()

    # 批量调用
    print("\n=== batch ===")
    texts = ["单机性能的瓶颈在哪里？", "为什么需要分库分表？"]
    for translated in chain.batch([{"text": t} for t in texts]):
        print("-", translated)


if __name__ == "__main__":
    main()

7.4.2 示例二：RunnableParallel 拆分子任务再合并

一个输入文本，同时交给"摘要子链"和"关键词子链"两条分支并行处理，最后合并成一段报告。手写实现时你需要自己写并发代码（CompletableFuture 或线程池）并处理异常；LCEL 里这就是管道本身的能力。

import os

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda, RunnableParallel
from langchain_openai import ChatOpenAI


def merge(data: dict) -> str:
    """合并两条子链的结果。RunnableParallel 会把分支结果按 key 汇成 dict。"""
    return f"[摘要] {data['summary']}\n[关键词] {data['keywords']}"


def main():
    model = ChatOpenAI(
        model=os.getenv("LLM_MODEL", "deepseek-chat"),
        api_key=os.getenv("OPENAI_API_KEY"),
        base_url=os.getenv("OPENAI_BASE_URL", "https://api.deepseek.com/v1"),
    )

    # 子链 1：生成摘要（两链共享输入 {text}）
    summary_chain = (
        ChatPromptTemplate.from_template("用一句话概括下面内容：\n{text}")
        | model
        | StrOutputParser()
    )

    # 子链 2：提取关键词
    keywords_chain = (
        ChatPromptTemplate.from_template("提取下面内容的 3 个关键词，用逗号分隔：\n{text}")
        | model
        | StrOutputParser()
    )

    # 先并行、后合并：输入 dict 会同时进入两条分支
    chain = (
        RunnableParallel(summary=summary_chain, keywords=keywords_chain)
        | RunnableLambda(merge)
    )

    result = chain.invoke({"text": "LangChain 是一个用于构建 LLM 应用的开源框架，核心是可组合管道。"})
    print(result)


if __name__ == "__main__":
    main()

工程启示：RunnableParallel 天然适合"一次输入、多视角加工"的任务——中英对照、多语言摘要、同时做翻译与情感分析、把同一份文本同时送入"检索"与"直接回答"两条分支（RAG 的路由雏形）。

7.5 与第 5 章手写实现的对比

下表把第 5 章（参见第 5 章）手写的"最小 Agent"与 LangChain 0.3 放在同一张表里，供你选型时参考：

结论：单机实验、教学演示、需要精确定制请求体——手写（第 5 章）更合适；多集成、多模型、团队协作、Agent + RAG 的规模化系统——用框架。但无论选哪条路，底层原理都必须懂：HTTP 调用、消息格式、function calling、token 计费，这些不会因为用了框架就消失，只是被封装起来——封装越厚，越需要你主动揭开它。

7.6 记忆 Memory：0.3 里推荐的做法

先给结论，再给证据：

LangChain 0.3 官方明确建议：新代码里不要再用 ConversationBufferMemory 这类 Memory 类管理历史，而是显式地把历史 messages 带进每次调用。 旧的 Memory 类属于 0.1 时代的遗留 API，在 0.3 里仍可用但已标记 deprecated，1.0 中已被移除。若你需要"框架自动管理会话历史"，请用 RunnableWithMessageHistory。

7.6.1 做法一：ConversationBufferMemory（旧式，不推荐）

import os

from langchain.memory import ConversationBufferMemory
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model=os.getenv("LLM_MODEL", "deepseek-chat"),
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.deepseek.com/v1"),
)

memory = ConversationBufferMemory()  # 默认把历史拼成一整段字符串
prompt = ChatPromptTemplate.from_template(
    "以下是之前的对话：\n{history}\n\n用户现在说：{input}"
)


def chat_legacy(user_input: str) -> str:
    # 手动取历史、手动拼进提示词、再手动保存——这就是"旧式"的别扭所在
    history = memory.load_memory_variables({})["history"]
    reply = model.invoke(prompt.invoke({"history": history, "input": user_input})).content
    memory.save_context({"input": user_input}, {"output": reply})
    return reply


if __name__ == "__main__":
    print(chat_legacy("我叫小王。"))
    print(chat_legacy("我叫什么名字？"))

问题很明显：历史与 prompt 的拼装逻辑散落在业务代码里，save_context / load_memory_variables 的键名约定晦涩，而且一旦想切换为流式或异步，这套东西完全帮不上忙。

7.6.2 做法二：显式 messages 管理（0.3 推荐）

import os

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model=os.getenv("LLM_MODEL", "deepseek-chat"),
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.deepseek.com/v1"),
)

# 会话历史容器：就是一个有序的消息列表，你可以用自己的存储替换它
history = InMemoryChatMessageHistory()


def chat_explicit(user_input: str) -> str:
    # 1) 用户输入先进历史
    history.add_user_message(user_input)
    # 2) 显式拼出本次请求的完整消息列表（system 每次都在最前）
    messages = [("system", "你是一个友好的助手。"), *history.messages]
    # 3) 调用模型
    reply = model.invoke(messages).content
    # 4) 助手回复也进历史
    history.add_ai_message(reply)
    return reply


if __name__ == "__main__":
    print(chat_explicit("我叫小王。"))
    print(chat_explicit("我叫什么名字？"))

这段代码没有任何魔法：历史就是一个 list[BaseMessage]，你可以随时打印、截断、持久化到数据库，也可以和 LCEL 无缝衔接（比如把 messages 直接喂给 prompt | model 链）。

7.6.3 对比与建议

如果你的场景是"多用户多会话、希望框架自动按 session_id 维护历史"，0.3 提供了一个折中方案 RunnableWithMessageHistory，它把链与按 session 存储的历史自动接起来（配合 MessagesPlaceholder 使用）：

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的助手。"),
    MessagesPlaceholder("history"),          # 占位：框架会在这里注入历史
    ("user", "{input}"),
])

sessions: dict[str, InMemoryChatMessageHistory] = {}


def get_history(session_id: str) -> InMemoryChatMessageHistory:
    if session_id not in sessions:
        sessions[session_id] = InMemoryChatMessageHistory()
    return sessions[session_id]


chain_with_history = RunnableWithMessageHistory(
    prompt | model,
    get_history,
    input_messages_key="input",
    history_messages_key="history",
)

reply = chain_with_history.invoke(
    {"input": "我叫小王。"},
    config={"configurable": {"session_id": "user-001"}},
)

它适合生产环境（session 维度隔离、自动读写历史），但学习阶段建议先吃透显式 messages 的做法——理解了它，再看 RunnableWithMessageHistory 只是"自动拼消息"的封装，并不神秘。

7.7 流式与异步

流式：chain.stream(...) 以增量 chunk 返回输出，适合打字机效果的 WebSSE 推送；异步：.ainvoke / .astream / .abatch 是基于 asyncio 的非阻塞版本，适合高并发 Web 服务（如用 FastAPI 承载 Agent 服务时）。

一句话：stream 管"边生成边消费"，async 管"不阻塞事件循环"，二者可叠加为 .astream()。

import asyncio

# 异步流式：在异步上下文中边生成边消费
async def async_example():
    chain = ...  # 任意 LCEL 链，复用同一管道定义
    async for chunk in chain.astream({"text": "用一句话解释流式输出。"}):
        print(chunk, end="", flush=True)
    print()


if __name__ == "__main__":
    asyncio.run(async_example())

工程上记住一点：流式/异步不是链的额外功能，而是 Runnable 协议的一部分。你手写实现时"为流式写的分支逻辑"在 LCEL 里从组件层面就统一解决了——这也是第 8 章做 Agent 流式输出、第 9-10 章在 LangGraph 里做异步编排的基础。

小结

1. LangChain 是可组合的工具箱：六个核心抽象（Model / Prompt / OutputParser / Retriever / Memory / Tool）各自只做一件事，通过 Runnable 协议统一组合。
2. LCEL 是 LangChain 的骨架：A | B 等于顺序管道，RunnableParallel 是分支并行，RunnableLambda 让你把任意 Python 函数织进管道，RunnablePassthrough 负责透传与补字段。
3. 统一执行协议（invoke / stream / batch / 异步三兄弟）让同一个链天然支持单次、流式、批量、异步四种用法。
4. 0.3 的记忆推荐显式 messages 管理；ConversationBufferMemory 是遗留 API，新代码不碰。
5. 手写（第 5 章）与框架并不对立：小系统手写、大系统用框架，但必须理解底层。
6. 本章是第 8 章（工具调用 / RAG）与第 9-10 章（LangGraph）的基石：工具是 Tool 抽象 + function calling，LangGraph 本身也是建立在 Runnable 之上的状态机。

练习

1. 修改 7.4.1 的翻译链，把 StrOutputParser 换成 JsonOutputParser，让模型输出 {"translation": "..."} 的结构化结果（提示：在 prompt 里明确要求输出 JSON）。
2. 用 RunnableParallel 把同一文本同时做"翻译成英文"与"提取关键词"两个子任务，再合并输出，体会分支并行的写法。
3. 把 7.6.2 的显式 messages 历史替换成你自己的持久化（比如写到一个 JSON 文件），实现"重启进程后历史仍在"。
4. 对照第 5 章手写实现，分别用手写方式与 LangChain 实现同一翻译链，打印两者的请求体与耗时，比较差异（可用 get_openai_callback() 查看 token 统计）。
5. 挑战：用 RunnableLambda 包装一个"调用你们内部 Java 服务"的函数，把它插入翻译链的 model 之后，实现"先翻译、再调内部服务校验"的管道。

参考答案

<details>
<summary>练习1（StrOutputParser 换 JsonOutputParser）</summary>

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_template(
    """把下面的英文翻译成中文，输出 JSON：
    {{"translation": "..."}}

    原文：{text}"""
)

chain = prompt | llm | JsonOutputParser()
result = chain.invoke({"text": "Hello, world"})
print(result)            # {'translation': '你好，世界'}

要点：

• 提示里必须显式要求输出 JSON（并给出结构），否则模型可能输出纯文本，JsonOutputParser 解析失败会抛异常；
• JsonOutputParser 返回的是已解析的 dict（不再是字符串），后续链可以直接 result["translation"]；
• 与 with_structured_output 的区别：JsonOutputParser 不校验字段，只是"解析 + 修复 JSON 格式"；要求字段强约束请用 Pydantic（第 3.6 节）。
  </details>

<details>
<summary>练习2（RunnableParallel 分支并行）</summary>

from langchain_core.runnables import RunnableParallel

translate = ChatPromptTemplate.from_template(
    "翻译成英文：{text}") | llm | StrOutputParser()
keywords = ChatPromptTemplate.from_template(
    "提取关键词，用逗号分隔：{text}") | llm | StrOutputParser()

parallel = RunnableParallel(
    translation=translate,
    keywords=keywords,
)
result = parallel.invoke({"text": "航班因天气取消"})
# {'translation': 'The flight was cancelled due to weather.',
#  'keywords': '航班,天气,取消'}

要点：

• 分支的输入字典共享 {"text": ...}，每个分支独立执行，互不等待、并行发出；
• 合并输出后，可以继续 | prompt2 | llm 做"汇总/再加工"——这是"子任务并行 + 结果汇聚"的通用写法；
• 若两个分支调的是同一家 API，注意并发上限与限流（第 12 章），可调小并发或分批。
  </details>

<details>
<summary>练习3（messages 历史持久化到 JSON 文件）</summary>

把 7.6.2 的显式 messages 序列化落盘：

import json
from langchain_core.messages import messages_from_dict, messages_to_dict

class JsonFileMemory:
    def __init__(self, path: str):
        self.path = path

    def load(self) -> list:
        try:
            with open(self.path, encoding="utf-8") as f:
                return messages_from_dict(json.load(f))
        except FileNotFoundError:
            return []

    def add(self, message) -> None:
        history = self.load() + [message]
        with open(self.path, "w", encoding="utf-8") as f:
            json.dump(messages_to_dict(history), f, ensure_ascii=False, indent=2)

# 使用：重启进程后历史仍在
mem = JsonFileMemory("history.json")
history = mem.load()
# ... 构造 SystemMessage + history + HumanMessage 请求
mem.add(ai_message)

要点：

• messages_to_dict / messages_from_dict 是 LangChain 官方的消息序列化工具，各类消息（Human/AI/Tool/System）都能还原；
• 会话隔离：文件名带上 session_id（如 history_{session_id}.json），两个用户互不串台；
• 长度控制：加载时只取最近 N 条，否则历史无限膨胀，token 成本失控（第 2 章练习 4 的量级问题）。
  </details>

<details>
<summary>练习4（手写 vs LangChain 对比）</summary>

from langchain.callbacks import get_openai_callback

# 方式一：手写（第 5 章风格）——自己拼 prompt、自己调 API、自己解析
# 方式二：LangChain 链
with get_openai_callback() as cb:
    chain.invoke({"text": "..."})
    print(cb.total_tokens, cb.prompt_tokens, cb.completion_tokens, cb.total_cost)

对比维度：

• 请求体：本质相同——都是"System/User 消息 + 模型参数"，LangChain 只是帮你生成；打印 chain 或 trace 里的 prompt 可看到与手写版一致的结构；
• 耗时：链路本身几乎无差异（框架开销微秒级），差异来自模型调用本身；
• 真正的差异在工程侧：手写版要自己维护"重试、流式、批量、错误归一化"，LangChain 已内置；代价是框架抽象层——你写的链的行为与 with_structured_output 等组件的实现细节耦合，出问题时排查要理解框架内部。

结论：小规模原型手写完全够用且最透明；规模化后用框架省的是"体力活"，但底层原理（第 4、5 章）决定你能不能用得对。
</details>

<details>
<summary>练习5（RunnableLambda 接内部 Java 服务）</summary>

import requests
from langchain_core.runnables import RunnableLambda

def call_java_validation(translation: str) -> str:
    # 调用你们内部 Java 服务的校验接口（业务校验：术语、敏感词、规范）
    resp = requests.post(
        "http://internal-validator:8080/check",
        json={"text": translation},
        timeout=3,
    )
    resp.raise_for_status()
    return resp.json()["result"]          # 校验后的文本或错误信息

chain = (prompt | llm | StrOutputParser()
         | RunnableLambda(call_java_validation))

result = chain.invoke({"text": "..."})

要点：

• RunnableLambda 把任意 Python 函数（内部就是 HTTP 调用）织进链，返回值成为下一个环节的输入——框架边界在这里，公司内部服务在这里接入；
• 工程细节：timeout 必须设（LLM 链是同步的，校验服务挂了不能拖垮整条链）；失败要 raise_for_status 让异常冒泡，或用 try 降级；
• 这套"模型输出 → 内部服务强校验"的组合，是"LLM 生成 + 程序兜底"安全架构（第 12 章）的标准落地点——模型产出不可信，最终校验在服务端做。
  </details>

延伸阅读

• LangChain 官方文档（0.3 版本分支）：https://python.langchain.com/
• LCEL 概念与 Runnable 协议：https://python.langchain.com/docs/concepts/lcel/
• Runnable 接口参考（invoke / batch / stream / 异步）：https://python.langchain.com/api_reference/core/runnables.html
• LangChain 1.0 迁移说明：1.0 已 GA，本书基于 0.3，具体 API 差异一律以官方文档为准，勿直接照搬网上 0.1 时代的旧示例。