LangChain Expression Language（LCEL）语法精讲

核心知识点讲解

什么是LCEL？

LangChain Expression Language (LCEL)是LangChain v0.1.0引入的一种声明式语法，用于构建和组合LangChain组件。它提供了一种简洁、直观的方式来定义处理流程，使代码更加可读和可维护。

LCEL的基本概念

LCEL基于以下核心概念：

Runnables：可运行的组件，如模型、提示词模板、解析器等
操作符：用于组合Runnables的符号，如|（管道）、&（并行）等
链：通过操作符组合Runnables形成的处理流程

基本操作符

LCEL提供了多种操作符来组合组件：

操作符	功能	示例
`	`	管道：将前一个组件的输出作为后一个组件的输入
`&`	并行：同时执行多个组件，将结果合并	`(prompt1
`	>`	映射：将函数应用于输入
`if/else`	条件：根据条件选择不同的组件	`runnable1 if condition else runnable2`

实用案例分析

案例1：基本管道操作

from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI
from langchain.output_parsers import ResponseSchema, StructuredOutputParser

# 1. 定义响应模式
response_schemas = [
    ResponseSchema(name="answer", description="问题的回答"),
    ResponseSchema(name="confidence", description="答案的置信度，0-100之间的数字")
]

# 2. 创建输出解析器
output_parser = StructuredOutputParser.from_response_schemas(response_schemas)

# 3. 创建提示词模板
prompt = PromptTemplate(
    template="请回答以下问题，并提供置信度（0-100）：\n\n问题：{question}\n\n{format_instructions}",
    input_variables=["question"],
    partial_variables={"format_instructions": output_parser.get_format_instructions()}
)

# 4. 初始化语言模型
llm = OpenAI(temperature=0.7)

# 5. 使用LCEL构建链
chain = prompt | llm | output_parser

# 6. 运行链
result = chain.invoke({"question": "什么是人工智能？"})
print(f"回答: {result['answer']}")
print(f"置信度: {result['confidence']}")

案例2：并行执行

from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI

# 1. 创建提示词模板
prompt1 = PromptTemplate(
    template="请用中文回答：{question}",
    input_variables=["question"]
)

prompt2 = PromptTemplate(
    template="请用英文回答：{question}",
    input_variables=["question"]
)

# 2. 初始化语言模型
llm = OpenAI(temperature=0.7)

# 3. 创建并行链
parallel_chain = (prompt1 | llm) & (prompt2 | llm)

# 4. 运行链
result = parallel_chain.invoke({"question": "什么是人工智能？"})
print(f"中文回答: {result[0]}")
print(f"英文回答: {result[1]}")

案例3：使用映射操作符

from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI

# 1. 创建提示词模板
prompt = PromptTemplate(
    template="请总结以下文本：\n\n{text}",
    input_variables=["text"]
)

# 2. 初始化语言模型
llm = OpenAI(temperature=0.7)

# 3. 创建链，使用映射操作符处理输入
chain = prompt | llm | (lambda x: {"summary": x.strip()})

# 4. 运行链
text = "人工智能（Artificial Intelligence，简称AI）是研究、开发用于模拟、延伸和扩展人的智能的理论、方法、技术及应用系统的一门新的技术科学。人工智能的发展历史可以分为几个阶段：1956年达特茅斯会议标志着人工智能的诞生；1970年代末到1980年代初，专家系统兴起；1990年代，机器学习技术取得突破；2010年代以来，深度学习技术的发展推动了人工智能的快速进步。"

result = chain.invoke({"text": text})
print(f"总结: {result['summary']}")

案例4：条件分支

from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI
from langchain.chains import RunnableBranch

# 1. 创建不同类型的提示词模板
math_prompt = PromptTemplate(
    template="请解决以下数学问题：{question}",
    input_variables=["question"]
)

general_prompt = PromptTemplate(
    template="请回答以下问题：{question}",
    input_variables=["question"]
)

# 2. 初始化语言模型
llm = OpenAI(temperature=0.7)

# 3. 创建条件分支链
def is_math_question(question):
    math_keywords = ["加", "减", "乘", "除", "等于", "计算", "数学", "公式"]
    return any(keyword in question for keyword in math_keywords)

branch = RunnableBranch(
    (is_math_question, math_prompt | llm),
    general_prompt | llm
)

# 4. 运行链
result1 = branch.invoke({"question": "1+1等于多少？"})
print(f"数学问题回答: {result1}")

result2 = branch.invoke({"question": "什么是人工智能？"})
print(f"一般问题回答: {result2}")

代码解析

案例1：基本管道操作

组件创建：创建了提示词模板、语言模型和输出解析器
LCEL链构建：使用|操作符将组件连接成链
运行链：使用invoke方法传入输入，获取解析后的结果

案例2：并行执行

并行链构建：使用&操作符同时执行两个链（中文回答和英文回答）
结果合并：并行执行的结果会以元组的形式返回

案例3：使用映射操作符

映射函数：使用|>操作符和lambda函数处理模型输出
结果转换：将模型的原始输出转换为结构化的字典

案例4：条件分支

条件函数：定义了一个函数来判断问题是否为数学问题
分支链构建：使用RunnableBranch创建条件分支链
动态选择：根据输入的类型选择不同的处理流程

高级技巧

1. 处理列表输入

使用map方法处理列表输入：

from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI

prompt = PromptTemplate(
    template="请总结以下文本：\n\n{text}",
    input_variables=["text"]
)

llm = OpenAI(temperature=0.7)

chain = prompt | llm

# 处理列表输入
texts = [
    "人工智能是研究、开发用于模拟、延伸和扩展人的智能的理论、方法、技术及应用系统的一门新的技术科学。",
    "机器学习是人工智能的一个分支，致力于研究如何使计算机从数据中学习并改进性能。",
    "深度学习是机器学习的一个分支，使用多层神经网络来模拟人脑的学习过程。"
]

results = chain.batch([{"text": text} for text in texts])

for i, result in enumerate(results):
    print(f"总结{i+1}: {result}")

2. 自定义Runnables

创建自定义的Runnable组件：

from langchain.schema.runnable import Runnable
from typing import Dict, Any

class CustomRunnable(Runnable):
    def invoke(self, input: Dict[str, Any], config=None) -> Dict[str, Any]:
        # 自定义处理逻辑
        text = input["text"]
        return {
            "original": text,
            "processed": text.upper(),
            "length": len(text)
        }

# 使用自定义Runnable
custom_runnable = CustomRunnable()
result = custom_runnable.invoke({"text": "hello world"})
print(result)

# 与其他组件组合
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI

prompt = PromptTemplate(
    template="处理后的文本：{processed}\n请对其进行评论：",
    input_variables=["processed"]
)

llm = OpenAI(temperature=0.7)

chain = custom_runnable | prompt | llm
result = chain.invoke({"text": "hello world"})
print(result)

3. 使用配置参数

通过配置参数控制组件行为：

from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI

prompt = PromptTemplate(
    template="请回答：{question}",
    input_variables=["question"]
)

llm = OpenAI(temperature=0.7)

chain = prompt | llm

# 使用不同的配置
result1 = chain.invoke(
    {"question": "什么是人工智能？"}, 
    config={"max_tokens": 100}
)
print(f"简短回答: {result1}")

result2 = chain.invoke(
    {"question": "什么是人工智能？"}, 
    config={"max_tokens": 500}
)
print(f"详细回答: {result2}")

4. 错误处理

使用RunnablePassthrough和try/except处理错误：

from langchain.schema.runnable import RunnablePassthrough
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI

def safe_process(inputs):
    try:
        # 可能会出错的处理
        return {"result": inputs["text"].upper()}
    except Exception as e:
        return {"error": str(e)}

prompt = PromptTemplate(
    template="请处理：{result}",
    input_variables=["result"]
)

llm = OpenAI(temperature=0.7)

chain = (
    RunnablePassthrough.assign(
        processed=lambda x: safe_process(x)
    )
    | prompt
    | llm
)

# 正常输入
result1 = chain.invoke({"text": "hello world"})
print(f"正常处理: {result1}")

# 错误输入
result2 = chain.invoke({"text": None})
print(f"错误处理: {result2}")

最佳实践

1. 代码组织

模块化：将复杂的链分解为小的、可重用的组件
命名：为Runnables和链使用清晰、描述性的名称
注释：为复杂的链添加注释，解释其功能和设计意图

2. 性能优化

缓存：对重复的计算使用缓存
并行：使用&操作符并行执行独立的组件
批处理：对多个输入使用batch方法批量处理

3. 调试技巧

日志：使用verbose=True启用详细日志
中间结果：使用RunnablePassthrough检查中间结果
分段测试：单独测试每个组件，确保它们正常工作

总结

LangChain Expression Language (LCEL)提供了一种强大、直观的方式来构建和组合LangChain组件。通过本集的学习，我们掌握了：

基本操作符：使用|、&等操作符组合组件
并行执行：使用&操作符同时执行多个组件
条件分支：使用RunnableBranch根据条件选择不同的处理流程
列表处理：使用batch方法处理多个输入
自定义Runnables：创建符合特定需求的自定义组件
错误处理：优雅地处理执行过程中的错误

LCEL的简洁语法使代码更加可读和可维护，同时提供了足够的灵活性来处理复杂的场景。它是构建LangChain应用的重要工具，掌握LCEL将大大提高你的开发效率。

课后思考

什么时候应该使用LCEL而不是传统的Chain类？
如何处理LCEL链中的复杂依赖关系？
如何优化LCEL链的性能？
如何调试复杂的LCEL链？

在下一集中，我们将学习什么是ReAct（Reason + Act）范式，以及如何使用它来构建更智能的AI系统。