LangChain 链式调用深度解析：从 LLMChain 到 RouterChain 的全景图

LangChain 的核心价值在于将 LLM 应用抽象为“链”(Chain)，实现组件的可编排、可复用与可维护。从最简单的 LLMChain 到复杂的条件路由 RouterChain，再到新一代 LCEL 表达式，理解这些链的设计思想与运作机制，是驾驭 LangChain 的关键。以下逐一深度剖析。

LLMChain 是 LangChain 最简单的链，它由哪些组件构成？写出它的核心执行流程。

LLMChain 是 LangChain 中最基础、最经典的链。虽然目前官方已推荐使用更简洁的 LCEL 表达式，但 LLMChain 仍是理解链式调用思想的绝佳起点。

🧩 核心组件：

一个标准的 LLMChain 由三个核心组件构成：

• PromptTemplate（提示模板）：负责将用户输入或其他变量格式化为一个结构化的 Prompt 字符串或消息列表。

• LLM/ChatModel（语言模型）：接收 Prompt 并生成文本响应，可以是文本补全模型或聊天模型。

• （可选）OutputParser（输出解析器）：对模型输出进行解析和结构化处理，将纯文本转换为程序可用的格式。

⚙️ 核心执行流程： 当调用 chain.invoke(input_dict) 时，LLMChain 内部执行以下步骤：

1. Prompt 格式化：将传入的键值对字典（如 {"topic": "量子计算"}）传递给 PromptTemplate，根据模板中的占位符生成最终的 Prompt 字符串或消息列表。

2. 模型调用：将格式化后的 Prompt 传递给 LLM/ChatModel，调用模型 API 并等待模型生成响应文本。

3. （可选）输出解析：如果配置了 OutputParser，将模型生成的原始文本传递给解析器，解析器按照预定义规则提取结构化数据。

4. 返回结果：将最终的响应（解析后的结构化数据或原始文本）返回给调用方。

📜 代码示例：

from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain.chains import LLMChain

prompt = PromptTemplate.from_template("请用{language}写一段关于{topic}的介绍。")
model = ChatOpenAI(model="gpt-4o", temperature=0.7)

chain = LLMChain(prompt=prompt, llm=model)
result = chain.invoke({"language": "中文", "topic": "机器学习"})
# 等价于
# chain = prompt | model

📝 内部伪代码：

class LLMChain:
    def __init__(self, prompt, llm, output_parser=None):
        self.prompt = prompt
        self.llm = llm
        self.output_parser = output_parser

    def invoke(self, inputs):
        formatted_prompt = self.prompt.format(**inputs)
        response = self.llm.invoke(formatted_prompt)
        if self.output_parser:
            return self.output_parser.parse(response)
        return response

虽然 LLMChain 的实现很简单，但它清晰地体现了 LangChain 的核心思想：Prompt + Model + OutputParser 的组合模式，为更复杂的链打下了基础。

如何使用 LCEL 复刻一个 LLMChain？写出等价的 LCEL 表达式。

LCEL（LangChain Expression Language）是 LangChain 的新一代编程范式，使用 | 管道操作符串联组件。用 LCEL 复刻 LLMChain 非常直观，能清晰表达数据流动。

📜 等价的 LCEL 表达式：

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

# 1. 定义 Prompt（使用消息模板更符合现代实践）
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个用{language}回答问题的助手。"),
    ("human", "{topic}")
])

# 2. 初始化模型
model = ChatOpenAI(model="gpt-4o", temperature=0.7)

# 3. 初始化输出解析器（可选）
parser = StrOutputParser()

# 4. LCEL 表达式 = LLMChain 的完全等价形式
chain = prompt | model | parser

# 5. 调用
result = chain.invoke({"language": "中文", "topic": "深度学习"})

🔍 与 LLMChain 的对比：

查看内嵌表格

💡 LCEL 的优势：

• 数据从左到右流动，直观清晰；

• 原生支持流式、异步、并行、回退等高级特性；

• 链本身也是 Runnable，可无限嵌套组合。

因此，所有新项目都应优先使用 LCEL 表达式，旧版 LLMChain 主要用于理解基本原理和维护遗留代码。

解释 SimpleSequentialChain 和 SequentialChain 的区别，以及各自的适用场景。

当我们需要将多个链串联，形成更复杂的处理流程时，就会用到 SequentialChain（顺序链）。LangChain 提供了两种顺序链。

🔗 SimpleSequentialChain：

• 最简单的顺序链，将多个子链首尾相连。

• 数据传递规则：上一个链的单一输出，直接作为下一个链的唯一输入。

• 局限性：要求每个子链只有一个输入变量和一个输出变量，无法处理多变量传递。

🔗 SequentialChain：

• 更强大、更灵活的顺序链。

• 数据传递规则：允许显式指定每个子链的输入变量和输出变量。可以组合多个变量，将前面链的多个输出和外部输入一起传递给后面的链。

• 核心能力：解决了多变量传递和复杂输入输出映射问题。

📊 对比总结：

查看内嵌表格

🎯 适用场景：

• SimpleSequentialChain：当整个流程是线性的，且上下游之间只有一种数据传递关系时使用。例如，输入主题→生成文章→翻译成英文。

• SequentialChain：当需要从上游链提取多个输出，或需要将全局参数注入下游链时使用。例如，在合同分析中，先提取当事人信息，再结合“风险偏好”参数进行风险评估。

⚠️ 注意：这两个类属于旧版 Chain API，在 LCEL 中可以通过 RunnableSequence 和 RunnableParallel 更优雅地实现类似功能。

在一个 SequentialChain 中，如何将前面链的输出传递给后面的链作为特定变量？

在 SequentialChain 中，数据传递的关键在于显式声明 input_variables 和 output_variables。

📜 示例场景：

• 链1：根据剧本生成影评（输出 review）。

• 链2：将影评翻译成英文（输入 text，输出 translated）。

• 链3：提取英文影评中的情感关键词（输入 english_review，输出 keywords）。

from langchain.chains import LLMChain, SequentialChain
from langchain.prompts import PromptTemplate

# 链1：生成中文影评
prompt1 = PromptTemplate.from_template("为电影《{movie}》写一段简短的影评。")
chain1 = LLMChain(prompt=prompt1, llm=model, output_key="review")

# 链2：翻译成英文，注意输入变量是 'text'
prompt2 = PromptTemplate.from_template("将以下中文翻译成英文：{text}")
chain2 = LLMChain(prompt=prompt2, llm=model, output_key="translated")

# 链3：提取关键词，输入变量需要和上级输出匹配
prompt3 = PromptTemplate.from_template("从以下英文影评中提取3个情感关键词：{english_review}")
chain3 = LLMChain(prompt=prompt3, llm=model, output_key="keywords")

# 组装 SequentialChain
overall_chain = SequentialChain(
    chains=[chain1, chain2, chain3],
    input_variables=["movie"],  # 初始输入变量
    output_variables=["review", "translated", "keywords"],  # 最终输出变量
    verbose=True
)
result = overall_chain({"movie": "流浪地球"})

🔧 关键机制：

• 链2 的 Prompt 模板中有占位符 {text}，SequentialChain 会自动将上游 chain1 的输出（键为 review）传递过来。变量名匹配是自动的，前提是上游的 output_key 必须和下游 Prompt 中的占位符名称一致。如果名称不一致，可以通过中间变量重命名。

• 链3 同理，Prompt 中的 {english_review} 会自动匹配 chain2 的输出键 translated。

⚙️ 传递多个变量：

如果下游链需要多个输入（例如，既需要上游输出，也需要外部参数），只需在 Prompt 中声明多个占位符，SequentialChain 会自动从已知变量中匹配。上游输出键和外部输入键会自动合并到可用变量池中。

如果你有多个子链，每个链的输出是下一个链的输入，但需要额外注入其他参数，你怎么做？

在纯 SequentialChain 中，所有变量默认来自上游输出或初始输入。如果需要额外注入不属于任何上游输出的外部参数，可以通过以下方式实现。

🔧 方法一：在初始输入中注入 将所有需要的额外参数作为 input_variables 的一部分在调用时传入。SequentialChain 会将其保留在整个变量池中，下游链可以直接引用。

# 假设 chain2 需要外部参数 'style'
prompt2 = PromptTemplate.from_template("将以下内容翻译成英文，风格为{style}：{text}")
chain2 = LLMChain(prompt=prompt2, llm=model, output_key="translated")

overall_chain = SequentialChain(
    chains=[chain1, chain2],
    input_variables=["movie", "style"],  # 额外参数在这里声明
    output_variables=["translated"]
)

# 调用时注入额外参数
result = overall_chain({"movie": "流浪地球", "style": "幽默风趣"})

🔧 方法二：使用 RunnablePassthrough 在 LCEL 中注入 在 LCEL 中，RunnablePassthrough 是专门用于数据透传和额外参数注入的工具。

from langchain_core.runnables import RunnablePassthrough

chain1 = prompt1 | model1 | StrOutputParser()
chain2 = prompt2 | model2 | StrOutputParser()

# 额外参数注入：将 'style' 参数透传，同时链1的输出作为 'text'
full_chain = (
    {"text": chain1, "style": RunnablePassthrough()}  # 透传 'style'
    | prompt2 | model2
)

result = full_chain.invoke({"movie": "...", "style": "幽默风趣"})

这种方式将链1的输出映射为 text，同时保持外部 style 参数不变，实现了额外的参数注入，比 SequentialChain 更灵活和直观。

RouterChain 是如何根据输入内容动态选择下游链的？它内部使用了什么组件？

RouterChain（路由链）解决了“根据用户意图将请求分发到不同处理链”的问题。它不是固定的线性流，而是动态条件路由。

🧩 内部组件：

• Router（路由器）：核心决策组件，负责分析输入并决定应该激活哪个下游链。可以是基于 LLM 的语义路由器，也可以是基于嵌入向量相似度的路由器。

• Destination Chains（目标链）：一组预定义的子链，每个子链负责处理一类特定任务（如翻译链、摘要链、问答链）。

• RouterOutputParser（路由输出解析器）：专门解析路由器 LLM 的原始输出，提取出目标链的名称和下一步的输入。

⚙️ 执行流程：

1. 输入分析：用户查询传入 RouterChain。

2. 路由决策：RouterChain 调用内部 Router（可能是一个 LLM），分析用户意图。Router 返回一个决策结果（如 {"destination": "translate_chain", "next_inputs": {"text": "你好世界"}}）。

3. 解析路由结果：RouterOutputParser 解析 Router 的原始输出，提取出目标链名称和传递参数。

4. 分发执行：RouterChain 根据解析出的目标链名称，从预定义的 Destination Chains 中选取对应的链，并将参数传递过去。

5. 返回结果：目标链执行并返回最终结果。

📜 代码示例：

from langchain.chains.router import LLMRouterChain, RouterOutputParser
from langchain.chains.router.multi_prompt import MultiPromptChain
from langchain.prompts import PromptTemplate

# 定义目标链的模板
translate_template = "将以下文本翻译成英文：{input}"
summary_template = "用一句话总结以下内容：{input}"

# 定义路由器的 Prompt
router_template = """
根据用户输入，选择下一步操作：
- 如果用户要求翻译，输出：{{"destination": "translate", "next_inputs": {{"input": "用户原文"}}}}
- 如果用户要求摘要，输出：{{"destination": "summary", "next_inputs": {{"input": "用户原文"}}}}
用户输入：{input}
"""
router_prompt = PromptTemplate.from_template(router_template)

# 创建路由器
router = LLMRouterChain.from_llm(llm=model, prompt=router_prompt)

# 创建目标链映射
destination_chains = {
    "translate": translate_chain,
    "summary": summary_chain,
}

# 创建 MultiPromptChain（一种 RouterChain 实现）
router_chain = MultiPromptChain(
    router_chain=router,
    destination_chains=destination_chains,
    default_chain=default_chain,
    verbose=True,
)

result = router_chain.invoke({"input": "你好世界，请翻译成英文"})

MultiPromptChain 是 RouterChain 的具体实现，它内部集成了 LLMRouterChain 和 RouterOutputParser，自动完成路由决策。

写一个 RouterChain 的示例：根据用户查询的类别（翻译、摘要、问答）路由到不同的 LLMChain。

这里展示一个完整的 RouterChain 实战示例，根据用户输入动态路由到三个不同的处理链。

from langchain.chains.router import MultiPromptChain
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o", temperature=0)

# 定义三个目标链的 Prompt
translate_prompt = PromptTemplate.from_template("将以下文本翻译成英文：{input}")
summary_prompt = PromptTemplate.from_template("用一句话总结以下内容：{input}")
qa_prompt = PromptTemplate.from_template("回答以下问题：{input}")

# 创建三个 LLMChain 实例
translate_chain = LLMChain(llm=model, prompt=translate_prompt, output_key="result")
summary_chain = LLMChain(llm=model, prompt=summary_prompt, output_key="result")
qa_chain = LLMChain(llm=model, prompt=qa_prompt, output_key="result")

# 定义路由器 Prompt
router_template = """
根据用户的输入，判断应该执行哪个操作：
- 如果用户要求翻译，输出：{{"destination": "translate", "next_inputs": {{"input": "用户要翻译的文本"}}}}
- 如果用户要求摘要，输出：{{"destination": "summary", "next_inputs": {{"input": "用户要摘要的文本"}}}}
- 如果用户问问题，输出：{{"destination": "qa", "next_inputs": {{"input": "用户的问题"}}}}

用户输入：{input}
请用 JSON 格式输出。
"""

router_prompt = PromptTemplate.from_template(router_template)

# 创建路由器链
from langchain.chains.router import LLMRouterChain
router = LLMRouterChain.from_llm(llm=model, prompt=router_prompt, verbose=True)

# 目标链映射
destination_chains = {
    "translate": translate_chain,
    "summary": summary_chain,
    "qa": qa_chain,
}

# 创建默认链（当路由无法匹配时使用）
default_chain = LLMChain(llm=model, prompt=PromptTemplate.from_template("回答：{input}"))

# 组装 RouterChain
router_chain = MultiPromptChain(
    router_chain=router,
    destination_chains=destination_chains,
    default_chain=default_chain,
    verbose=True,
)

# 测试
result1 = router_chain.invoke({"input": "请把'你好世界'翻译成英文"})
result2 = router_chain.invoke({"input": "写一篇关于AI的文章，然后用一句话总结"})

⚙️ 工作原理：

• LLMRouterChain 调用 LLM 分析用户输入，返回 JSON 格式的路由决策。

• RouterOutputParser 解析这个 JSON，得到 destination 和 next_inputs。

• MultiPromptChain 根据 destination 选择对应的目标链，将 next_inputs 传递过去并执行。

LLMRouterChain 和 EmbeddingRouterChain 分别适合什么场景？它们的路由策略有何不同？

🆚 LLMRouterChain：

• 路由策略：基于 LLM 的语义理解。将用户输入和候选目标描述一起发给 LLM，LLM 推理出最合适的目标链。

• 优点：极其灵活，能理解复杂、模糊的意图，无需预先为每个类别准备大量样本。

• 缺点：每次路由需要额外调用 LLM，增加延迟和成本；准确性依赖 LLM 自身能力；确定性较低。

• 适用场景：意图类别动态变化、模糊语义理解、复杂条件路由；例如客服系统中的多级意图分发。

🆚 EmbeddingRouterChain：

• 路由策略：基于向量嵌入相似度。预先为每个目标链的触发条件（如示例查询）计算嵌入向量，存入向量库。用户查询来时，计算其嵌入向量，通过向量相似度搜索找到最接近的预定义类别。

• 优点：速度快（无需额外 LLM 调用），成本低，结果可复现、确定性高。

• 缺点：灵活性受限，难以处理语义差距大的表述；需要预先准备足够的示例。

• 适用场景：意图类别明确且固定、高吞吐量、低延迟、需要可复现结果的场景；例如简单的命令分类、FAQ 路由。

📊 对比总结：

查看内嵌表格

在实际应用中，可以根据场景组合使用，或者先用 EmbeddingRouterChain 快速筛选，再用 LLMRouterChain 对模糊情况进行二次判断。

在 LCEL 中，如何实现条件路由？它和 RouterChain 的设计思路一致吗？

在 LCEL 中，我们使用 RunnableBranch（或 RunnableLambda）来实现条件路由，其设计思想与 RouterChain 完全一致，但实现更简洁、灵活。

📜 示例：根据用户查询类别路由到不同链

from langchain_core.runnables import RunnableBranch, RunnableLambda
from langchain_core.prompts import ChatPromptTemplate

# 1. 定义分类函数
def classify(input_dict):
    query = input_dict["query"]
    if "翻译" in query:
        return "translate"
    elif "摘要" in query:
        return "summary"
    elif "问答" in query or "?" in query:
        return "qa"
    else:
        return "default"

# 2. 定义各分支的处理链
translate_chain = ChatPromptTemplate.from_template("翻译成英文：{query}") | model
summary_chain = ChatPromptTemplate.from_template("总结：{query}") | model
qa_chain = ChatPromptTemplate.from_template("回答问题：{query}") | model
default_chain = ChatPromptTemplate.from_template("回答：{query}") | model

# 3. 使用 RunnableBranch 实现路由
branch = RunnableBranch(
    (lambda x: classify(x) == "translate", translate_chain),
    (lambda x: classify(x) == "summary", summary_chain),
    (lambda x: classify(x) == "qa", qa_chain),
    default_chain  # 默认分支
)

# 4. 调用
result = branch.invoke({"query": "请把'你好世界'翻译成英文"})

🔀 更优雅的方式：使用路由 LLM

router_prompt = ChatPromptTemplate.from_template("""
根据用户输入，输出一个操作类别（只输出类别名）：
- translate：用户要求翻译
- summary：用户要求摘要
- qa：用户问问题

用户输入：{query}
类别：""")

router_chain = router_prompt | model | StrOutputParser()

def route_to_chain(input_dict):
    category = router_chain.invoke(input_dict)
    if category == "translate":
        return translate_chain.invoke(input_dict)
    elif category == "summary":
        return summary_chain.invoke(input_dict)
    elif category == "qa":
        return qa_chain.invoke(input_dict)
    else:
        return default_chain.invoke(input_dict)

full_chain = RunnableLambda(route_to_chain)

🔗 设计思路对比：

• 旧版 RouterChain：将路由逻辑封装在特定类中，通过配置文件或代码定义路由规则，组件化程度高但抽象层次多，代码量大。

• LCEL RunnableBranch：将路由规则直接表达为“条件-分支”对，与 Python 的 if-elif-else 逻辑本质相同，学习成本低。所有的链都是标准 Runnable，可任意组合、测试、复用。

🛠️ 结论：LCEL 的条件路由是 RouterChain 思想的现代化、轻量化实现，完美诠释了“可组合性”的设计哲学。推荐在项目中使用 LCEL 分支来实现条件路由。

如果你有一个路由链，但下游链数量很多（上百个），怎样优化路由性能？

当路由链下游挂载了上百个目标链时，传统的基于LLM的语义路由（LLMRouterChain）会成为性能瓶颈——每次路由需要一次LLM调用，延迟高且成本昂贵。优化方案从路由算法和系统架构两个维度展开。

1.1 使用基于嵌入的快速路由

核心思想：预先为每个目标链计算其“触发条件”的文本嵌入（例如，每个链的典型用户查询样本），存入向量数据库。当真实查询到来时，计算其嵌入向量，通过向量相似度检索最匹配的目标链。

from langchain.chains.router import EmbeddingRouterChain
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

# 为每个链准备示例查询
chain_examples = {
    "translate_chain": ["翻译成英文", "用英语怎么说", "请翻译"],
    "summary_chain": ["总结", "摘要", "概括"],
    "qa_chain": ["什么是", "如何", "为什么"],
    # ... 上百个链
}

# 创建嵌入路由器
router = EmbeddingRouterChain.from_names_and_descriptions(
    names_and_descriptions={name: " ".join(examples) for name, examples in chain_examples.items()},
    embedding=OpenAIEmbeddings(),
    vectorstore_cls=Chroma,
    top_k=1  # 选择最相似的一个
)

这种方案的延迟仅为向量检索时间（毫秒级），且不消耗LLM Token。

1.2 分层路由

如果链数量极大，单一向量检索可能不够精确。可以设计两级路由：

• 第一级：使用快速规则或小模型（如fastText）进行粗粒度分类，将查询分到某个“域”（如“客服”、“技术”、“销售”）。

• 第二级：在域内使用向量检索或轻量LLM进行细粒度匹配。 分层设计有效降低了单级路由的检索空间，提升了准确率和速度。

1.3 缓存路由决策

对于相似的重复查询，可以缓存其路由结果。实现方式：

• 使用内存缓存（如 functools.lru_cache）或在共享缓存（Redis）中存储 <query_hash, chain_name>。

• 结合语义缓存：如果新查询与某个已缓存查询的语义相似度超过阈值，直接复用路由决策，避免重复计算。

1.4 使用轻量级分类模型替代LLM

如果链的类别相对固定，可以训练一个轻量级文本分类器（如基于BERT的蒸馏模型、逻辑回归等），直接根据查询文本预测目标链。推理速度极快，成本几乎为零。这相当于把“路由”从LLM推理降级为传统模型推理。

1.5 在LCEL中动态路由

LCEL本身不提供高级路由器，但可以通过 RunnableLambda 自定义动态路由逻辑，在其中集成向量检索或分类器，达到与上述方案相同的效果。

from langchain_core.runnables import RunnableLambda

def dynamic_router(input_dict):
    query = input_dict["query"]
    # 调用向量检索，找到最佳链名
    best_chain_name = vector_search(query)  # 自己实现
    chain = chain_map[best_chain_name]
    return chain.invoke(input_dict)

router_chain = RunnableLambda(dynamic_router)

小结：优化路由性能的关键是减少或避免LLM调用，利用向量搜索、缓存、轻量模型等手段将路由决策成本降至最低。

如何监控每个链的调用次数和延迟？在 LangChain 中通常借助什么组件？

LangChain提供了强大的回调（Callback）机制，允许在链、LLM、工具等组件执行的关键节点插入自定义逻辑，是实现监控的基石。也可直接使用LangSmith一站式平台。

2.1 基于自定义回调的监控

创建一个继承 BaseCallbackHandler 的类，重写 on_chain_start 和 on_chain_end 方法，在方法中记录时间戳和链名称，然后聚合统计数据。

from langchain_core.callbacks import BaseCallbackHandler
import time
from collections import defaultdict

class ChainMonitorCallback(BaseCallbackHandler):
    def __init__(self):
        self.stats = defaultdict(lambda: {"count": 0, "total_latency": 0.0})
        self.start_times = {}

    def on_chain_start(self, serialized, inputs, *, run_id, **kwargs):
        chain_name = serialized.get("name", "unnamed_chain")
        self.start_times[run_id] = (chain_name, time.time())

    def on_chain_end(self, outputs, *, run_id, **kwargs):
        if run_id in self.start_times:
            chain_name, start = self.start_times.pop(run_id)
            latency = time.time() - start
            self.stats[chain_name]["count"] += 1
            self.stats[chain_name]["total_latency"] += latency

    def get_summary(self):
        summary = {}
        for name, data in self.stats.items():
            avg_latency = data["total_latency"] / data["count"] if data["count"] else 0
            summary[name] = {"calls": data["count"], "avg_latency_ms": avg_latency * 1000}
        return summary

使用方式：

monitor = ChainMonitorCallback()
chain.invoke(input, config={"callbacks": [monitor]})
print(monitor.get_summary())

可以将统计信息上报到Prometheus、Datadog等监控系统，或写入日志文件。

2.2 使用 LangSmith（推荐）

LangSmith是LangChain的官方可观测性平台。只需设置环境变量：

LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=your_key

所有链调用、LLM请求、工具执行都会被自动记录。平台提供调用次数、延迟、Token消耗、错误率等聚合图表，并支持按链名称筛选。这是生产环境中最便捷的监控方案。

解释 RunnableBranch 的作用，它与一连串的 | 管道相比有什么优势？

RunnableBranch 的作用

RunnableBranch 实现了基于条件的动态路由。它接受一个由 (condition, runnable) 对组成的列表，按顺序评估条件，将输入传递给第一个满足条件的Runnable执行。如果所有条件都不满足，则执行一个默认Runnable。

与 | 管道的对比

• 管道 (|)：表示固定的线性顺序。无论输入是什么，数据都必定依次流经A、B、C。它无法实现“根据内容选择不同处理路径”的逻辑。

• RunnableBranch：实现了“if-elif-else”的逻辑，让链的拓扑能够根据输入动态变化。

优势：

• 表达力更强：一条链即可处理多种情况，无需在外部编写条件判断并手动组装不同的链。

• 组件内聚：分支逻辑和对应的处理链封装在一起，代码更清晰。

• 可组合：RunnableBranch本身也是一个Runnable，可以继续与其他Runnable通过管道组合。

• 声明式：路由规则以列表形式声明，易于阅读和维护。

示例：

from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: "翻译" in x["query"], translate_chain),
    (lambda x: "摘要" in x["query"], summary_chain),
    qa_chain  # 默认
)

相比之下，如果用管道，你需要先判断，然后选择不同的链对象调用，逻辑与链的定义分离，不利于复杂应用的维护。

当链在执行过程中出现异常，LangChain 提供了哪些错误处理机制？（如 fallback、with_fallbacks）

LangChain提供了声明式回退和自定义异常捕获两种主要机制。

with_fallbacks

每个Runnable（包括链、模型、工具）都有一个 with_fallbacks 方法。它接收一个或多个备用Runnable，当主Runnable抛出异常时，自动依次尝试备用Runnable。

from langchain_openai import ChatOpenAI

# 主模型
gpt4 = ChatOpenAI(model="gpt-4o")
# 备用模型
gpt35 = ChatOpenAI(model="gpt-3.5-turbo")

robust_model = gpt4.with_fallbacks([gpt35])
chain = prompt | robust_model | parser

执行逻辑：如果 gpt4 因为网络超时、Rate Limit、API 500错误等抛出异常，robust_model 会自动捕获异常并调用 gpt35。如果 gpt35 也失败，异常会继续向上抛出。可以嵌套多个回退，例如：

gpt4.with_fallbacks([gpt35.with_fallbacks([local_llm])])

适用场景：

• 模型服务降级（主模型不可用时切换备用模型）。

• 工具调用降级（主工具失败时使用备用工具）。

• 跨云容灾（Azure OpenAI失败时切换到OpenAI）。

自定义异常捕获

如果需要在捕获异常后执行更复杂的逻辑（例如记录日志、清理状态、返回特定错误信息），可以使用 RunnableLambda 包裹一个带try/except的函数。

def safe_chain(input):
    try:
        return main_chain.invoke(input)
    except SpecificError as e:
        logger.error("Chain failed", error=e)
        return fallback_response

robust_chain = RunnableLambda(safe_chain)

这种方式更灵活，但失去了with_fallbacks的声明式简洁性。

如何为一条链设置全局 callback，以便记录所有中间步骤的日志？

可以通过在链调用时通过config参数注入回调实例来实现全局监控。这个回调会作用于链内部的所有子组件（LLM、工具、其他链）。

from langchain_core.callbacks import BaseCallbackHandler

class GlobalLoggerCallback(BaseCallbackHandler):
    def on_chain_start(self, serialized, inputs, *, run_id, parent_run_id=None, **kwargs):
        chain_name = serialized.get("name", "unnamed")
        print(f"[Chain Start] {chain_name} | Inputs: {inputs}")

    def on_chain_end(self, outputs, *, run_id, parent_run_id=None, **kwargs):
        print(f"[Chain End] run_id={run_id} | Outputs: {outputs}")

    def on_llm_start(self, serialized, prompts, *, run_id, parent_run_id=None, **kwargs):
        print(f"[LLM Start] prompts={prompts}")

    def on_llm_end(self, response, *, run_id, parent_run_id=None, **kwargs):
        print(f"[LLM End] response={response}")

logger = GlobalLoggerCallback()

# 调用时注入，影响整条链
chain.invoke(input_data, config={"callbacks": [logger]})

技巧：

• 可以将多个回调打包成一个列表传入，例如一个用于日志，一个用于监控。

• 在全局设置默认回调也可以，但不推荐，容易造成回调泄漏。更好的做法是在应用层统一管理回调的注入。

你如何在链中插入一个“不改变数据，只做日志记录”的步骤？用 LCEL 怎么实现？

这可以通过 RunnableLambda 包装一个原样返回输入的函数来实现。该函数内部完成日志记录。

from langchain_core.runnables import RunnableLambda
import logging

logger = logging.getLogger(__name__)

def log_step(data):
    logger.info(f"Intermediate data: {data}")
    return data  # 关键：原样返回输入，不修改

log_runnable = RunnableLambda(log_step)

chain = prompt | model | log_runnable | parser

log_runnable 被插入到管道中，接收上游输出，打印日志后将其透传给下游解析器，完全不影响数据流。

更优雅的方式：使用 RunnablePassthrough 的 side_effect 功能（部分版本支持），专门用于执行副作用操作。

LangChain 的 @chain 装饰器能做什么？它如何将普通函数转为 Runnable？

@chain 装饰器可以将一个普通的Python函数转换为Runnable，使该函数能直接使用 invoke, stream, batch 等标准接口，并能通过 | 与其他Runnable组合。

from langchain_core.runnables import chain

@chain
def my_custom_chain(inputs: dict) -> dict:
    query = inputs["query"]
    # 在此执行任意Python逻辑
    result = process_query(query)
    return {"output": result}

# 现在 my_custom_chain 就是一个 Runnable
chain = prompt | model | my_custom_chain | parser
result = chain.invoke({"query": "你好"})

本质：@chain 内部将函数包装为 RunnableLambda，并自动处理输入输出类型推断，使其符合Runnable协议。这使得开发者可以无缝地将现有业务逻辑融入LangChain的链式调用中。

在 LCEL 中，如何使用 RunnableLambda 将自定义 Python 逻辑整合到链中？

RunnableLambda 是连接自定义Python代码与LCEL管道的万能胶水。任何函数（同步或异步）包装后即可成为Runnable。

from langchain_core.runnables import RunnableLambda

def validate_and_clean(input_data: str) -> str:
    # 去除首尾空白，替换特定敏感词等
    cleaned = input_data.strip().replace("敏感词", "***")
    if len(cleaned) < 2:
        raise ValueError("Input too short")
    return cleaned

clean_runnable = RunnableLambda(validate_and_clean)

chain支持异步：如果函数是 async，可以用 RunnableLambda(func, afunc=async_func) 指定异步版本。
注意：RunnableLambda内部的函数输入和输出应该保持简单类型（如字符串、字典），因为这是链间数据交换的通用格式。
9. 如果将多个链并发执行而不是串行，应该用什么方式？写出代码示例。
使用 RunnableParallel（或 RunnableMap）可以实现多个链的并发执行。它接受一个字典，键为分支名称，值为一个Runnable。框架会利用线程池或异步协程同时执行所有分支，然后收集结果合并为一个字典输出。
示例：并发调用多个模型，然后汇总答案。 = clean_runnable | prompt | model | parser

支持异步：如果函数是 async，可以用 RunnableLambda(func, afunc=async_func) 指定异步版本。

注意：RunnableLambda内部的函数输入和输出应该保持简单类型（如字符串、字典），因为这是链间数据交换的通用格式。

如果将多个链并发执行而不是串行，应该用什么方式？写出代码示例。

使用 RunnableParallel（或 RunnableMap）可以实现多个链的并发执行。它接受一个字典，键为分支名称，值为一个Runnable。框架会利用线程池或异步协程同时执行所有分支，然后收集结果合并为一个字典输出。

示例：并发调用多个模型，然后汇总答案。

from langchain_core.runnables import RunnableParallel

# 假设有三个独立的链（或模型）
translate_chain = prompt_translate | model
summary_chain = prompt_summary | model
qa_chain = prompt_qa | model

# 并发执行
parallel = RunnableParallel(
    translation=translate_chain,
    summary=summary_chain,
    qa=qa_chain
)

# 调用
results = parallel.invoke({"input": "你好世界"})
print(results["translation"])  # 翻译结果
print(results["summary"])      # 摘要结果

实现细节：

• 在同步 invoke 中，RunnableParallel 默认使用线程池并发执行各分支。

• 在异步 ainvoke 中，使用 asyncio.gather 并发。

• 总耗时接近于最慢分支的耗时，大幅优于串行执行。

应用场景：多模型集成、多工具调用、同时进行检索和生成等。

什么是“链的配置化”？你是否尝试过将链的定义从代码中剥离到 YAML/JSON 配置文件？

链的配置化指将链的结构、Prompt模板、模型参数等从代码中分离，存储为外部配置文件（YAML、JSON），在运行时动态加载和组装。这实现了模板化和运维分离。

10.1 实现方式

LangChain支持通过 load_prompt, load_chain 等函数加载YAML/JSON文件。你可以将所有Prompt定义在YAML中，链的组装逻辑也可以写成配置文件（例如，定义链的类型和依赖组件）。

示例：prompts.yaml

translate_prompt:
  _type: chat
  input_variables: [text]
  messages:
    - role: system
      content: 你是一个翻译助手。
    - role: user
      content: "请将以下文本翻译成英文：{text}"

加载：

from langchain_core.prompts import load_prompt
prompt = load_prompt("prompts.yaml", "translate_prompt")

对于完整的链，可以使用langchain.chains.load_chain或直接通过配置文件定义链的序列，但在LCEL时代，更推荐将链的组装逻辑保留在代码中，因为LCEL表达式本身简洁且类型安全，而将可变部分（如Prompt、模型参数、工具列表）剥离到配置文件。

10.2 好处

• 非技术人员（如Prompt工程师）可独立修改提示词。

• 不同环境（开发/测试/生产）使用不同的配置。

• 版本控制与回滚更清晰。

在你看来，LangChain 中的 Chain 和普通 Python 函数的根本差异在哪里？什么时候该用链，什么时候该用函数？

根本差异

查看内嵌表格

何时用 Chain？

• 当应用包含多个LLM调用或LLM+工具+检索的复杂流程时。

• 需要生产级的监控、日志、容错时。

• 需要动态路由、并行执行、流式输出等高级功能时。

• 团队协作，希望复用标准化的组件。

何时用普通函数？

• 逻辑极其简单，只是单次API调用 + 简单字符串处理。

• 对性能有极致要求，不想引入任何框架开销。

• 快速原型验证，尚未确定长期架构。

• 核心逻辑与LLM无关的纯计算任务。

最佳实践：用 Chain 编排高层业务流程（“做什么”），用 RunnableLambda 包裹的普通函数处理自定义的数据清洗、格式转换等细节（“怎么做”）。这样既能享受框架的强大功能，又不失灵活性。

---