登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 安全 ›你应该懂的AI 大模型（五）之 LangChain 之 LCEL ...

你应该懂的AI 大模型（五）之 LangChain 之 LCEL

兼罔 2025-6-14 14:57:27
本文 对《LangChain》一文中的  Chain 与 LCEL 部分的示例进行详细的展示。
先回顾下 在LangChain框架中，Chain（链） 和 LCEL（LangChain Expression Language） 是两个密切相关但本质不同的概念。
Chain（链）: 是LangChain中处理流程的抽象概念，指将多个组件（模型、工具、逻辑）串联成一个可执行的任务序列。
LangChain Expression Language（LCEL）是一种声明式语言，可轻松组合不同的调用顺序构成 Chain。LCEL 自创立之初就被设计为能够支持将原型投入生产环境，无需代码更改，从最简单的“提示+LLM”链到最复杂的链（已有用户成功在生产环境中运行包含数百个步骤的 LCEL Chain）。
在本文中 LCEL 产生的对象，被叫做 runnable 或 chain，经常两种叫法混用。本质就是一个自定义调用流程。
1、Pipeline 式调用 PromptTemplate, LLM 和 OutputParser
  1. from langchain_openai import ChatOpenAI
  2. from langchain.prompts import ChatPromptTemplate
  3. from langchain_core.output_parsers import StrOutputParser
  4. from langchain_core.runnables import RunnablePassthrough
  5. from pydantic import BaseModel, Field, validator
  6. from typing import List, Dict, Optional
  7. from enum import Enum
  8. import json
复制代码
  1. # 输出结构
  2. class SortEnum(str, Enum):
  3.     data = 'data'
  4.     price = 'price'
  7. class OrderingEnum(str, Enum):
  8.     ascend = 'ascend'
  9.     descend = 'descend'
  12. class Semantics(BaseModel):
  13.     name: Optional[str] = Field(description="套餐名称", default=None)
  14.     price_lower: Optional[int] = Field(description="价格下限", default=None)
  15.     price_upper: Optional[int] = Field(description="价格上限", default=None)
  16.     data_lower: Optional[int] = Field(description="流量下限", default=None)
  17.     data_upper: Optional[int] = Field(description="流量上限", default=None)
  18.     sort_by: Optional[SortEnum] = Field(description="按价格或流量排序", default=None)
  19.     ordering: Optional[OrderingEnum] = Field(description="升序或降序排列", default=None)
  22. # Prompt 模板
  23. prompt = ChatPromptTemplate.from_messages(
  24.     [
  25.         ("system", "你是一个语义解析器。你的任务是将用户的输入解析成JSON表示。不要回答用户的问题。"),
  26.         ("human", "{text}"),
  27.     ]
  28. )
  30. # 模型
  31. llm = ChatOpenAI(model="gpt-4o", temperature=0)
  33. structured_llm = llm.with_structured_output(Semantics)
  35. # LCEL 表达式
  36. runnable = (
  37.     {"text": RunnablePassthrough()} | prompt | structured_llm
  38. )
  40. # 直接运行
  41. ret = runnable.invoke("不超过100元的流量大的套餐有哪些")
  42. print(
  43.     json.dumps(
  44.         ret.dict(),
  45.         indent = 4,
  46.         ensure_ascii=False
  47.     )
  48. )
复制代码
输出结果如下：
1.png

流式输出，就是那种一个字一个字蹦的输出，大家可以用下面代码体验一下玩一玩
  1. prompt = PromptTemplate.from_template("讲个关于{topic}的笑话")
  3. runnable = (
  4.     {"topic": RunnablePassthrough()} | prompt | llm | StrOutputParser()
  5. )
  7. # 流式输出
  8. for s in runnable.stream("李华"):
  9.     print(s, end="", flush=True)
复制代码
2、用LECL实现RAG

我们通过分割一个 pdf 文件形成向量库，示例如下：
  1. from langchain_openai import OpenAIEmbeddings
  2. from langchain_text_splitters import RecursiveCharacterTextSplitter
  3. from langchain_community.vectorstores import FAISS
  4. from langchain_openai import ChatOpenAI
  5. from langchain.chains import RetrievalQA
  6. from langchain_community.document_loaders import PyMuPDFLoader
  8. # 加载文档
  9. loader = PyMuPDFLoader("llama2.pdf")
  10. pages = loader.load_and_split()
  12. # 文档切分
  13. text_splitter = RecursiveCharacterTextSplitter(
  14.     chunk_size=300,
  15.     chunk_overlap=100,
  16.     length_function=len,
  17.     add_start_index=True,
  18. )
  20. texts = text_splitter.create_documents(
  21.     [page.page_content for page in pages[:4]]
  22. )
  24. # 灌库
  25. embeddings = OpenAIEmbeddings(model="text-embedding-ada-002")
  26. db = FAISS.from_documents(texts, embeddings)
  28. # 检索 top-2 结果
  29. retriever = db.as_retriever(search_kwargs={"k": 2})
复制代码
  1. from langchain.schema.output_parser import StrOutputParser
  2. from langchain.schema.runnable import RunnablePassthrough
  4. # Prompt模板
  5. template = """Answer the question based only on the following context:
  6. {context}
  8. Question: {question}
  9. """
  10. prompt = ChatPromptTemplate.from_template(template)
  12. # Chain
  13. rag_chain = (
  14.     {"question": RunnablePassthrough(), "context": retriever}
  15.     | prompt
  16.     | llm
  17.     | StrOutputParser()
  18. )
  20. rag_chain.invoke("Llama 2有多少参数")
复制代码
输出结果如下：
2.png

3、LECL 的意义是什么

在 LangChain 框架中，LCEL（LangChain Expression Language）的意义主要体现在增强模块间的解耦能力、提升流程编排的灵活性，并为复杂的 LLM（大语言模型）应用构建提供标准化的逻辑表达工具。以下是具体分析：
一、LangChain 中的模块解耦需求

LangChain 的核心目标是通过组合不同的组件（如 LLMs、Prompt 模板、工具调用、内存管理等）构建复杂的应用流程。例如：

  • 组件类型：LLM 调用、工具（如计算器、数据库查询）、内存（对话历史存储）、Prompt 模板、链（Chain，如 SequentialChain、RouterChain）等。
  • 解耦痛点：
    传统方式下，组件间的交互逻辑（如条件分支、数据转换、循环控制）可能硬编码在链的代码中，导致：

    • 链的逻辑难以复用或动态调整；
    • 组件更换（如从 OpenAI 切换到 Anthropic）时，需修改链的控制逻辑；
    • 复杂流程（如多轮工具调用、动态路由）的编排缺乏标准化表达。

二、LCEL 的核心意义：作为模块间的 “逻辑中介”

LCEL 是 LangChain 设计的一套声明式表达式语言，用于描述组件间的数据流动和逻辑控制。其核心价值体现在以下方面：
1. 分离逻辑控制与组件实现


  • 传统硬编码问题：
    例如，在链中通过 Python 代码实现条件判断：
    1. if result.score > 0.8:
    2.     return llm.predict(prompt.format(data=result))
    3. else:
    4.     return tool.query(data=result)
    复制代码
    此时，条件逻辑与链的代码强耦合，难以复用或动态修改。
  • LCEL 的解耦作用：
    将条件逻辑抽象为 LCEL 表达式，存储在配置中：
    1. condition: "{{ score > 0.8 }}"
    2. then:
    3.   action: llm.predict
    4.   params:
    5.     prompt: "用户输入：{{ input }}"
    6. else:
    7.   action: tool.query
    8.   params:
    9.     data: "{{ input }}"
    复制代码

    • 链的代码仅负责解析 LCEL 表达式，根据结果调用组件；
    • 逻辑变更时只需修改表达式，无需调整链的代码，实现 “逻辑外置”。

2. 标准化组件交互协议

LCEL 定义了一套统一的语法规则（如变量引用、运算符、函数调用），作为不同组件间的 “通信语言”：

  • 数据传递：通过 {{ variable }} 引用上游组件的输出（如 LLM 的返回结果、工具的查询数据）。
  • 逻辑操作：支持算术运算（+、-）、条件判断（if-else）、集合操作（map、filter）等，避免不同组件自定义逻辑语法。
  • 函数扩展：可注册自定义函数（如 format_date()、calculate_score()），供表达式调用，实现跨组件的逻辑复用。

示例：
通过 LCEL 表达式编排多步流程：
  1. from langchain.lcel import LCELChain
  3. expression = """
  4. let score = tool.calculate_score(input.data);
  5. if (score > 0.6) {
  6.   return llm.generate(prompt: "高分结果：{{ score }}");
  7. } else {
  8.   return tool.analyze(input.data);
  9. }
  10. """
  12. chain = LCELChain.from_expression(expression, components={
  13.   "tool": MyCustomTool(),
  14.   "llm": OpenAI(temperature=0.1)
  15. })
复制代码

  • 组件（tool、llm）仅需按约定输出数据字段，无需关心流程逻辑；
  • 链通过 LCEL 表达式动态调度组件，解耦组件与流程控制。
3. 支持动态流程编排与热更新

在 LangChain 中，复杂应用（如智能客服、数据分析助手）常需根据实时数据调整流程。LCEL 的声明式特性使其适合动态场景：

  • 场景 1：AB 测试
    通过后端配置不同的 LCEL 表达式，控制 A/B 组用户的流程分支（如优先调用工具 A 或工具 B），无需重启服务。
  • 场景 2：低代码平台
    用户通过可视化界面配置 LCEL 表达式（如拖拽生成 {{ order.amount > 1000 ? "大客户" : "普通客户" }}），快速生成定制化链，降低开发门槛。
  • 场景 3：实时策略调整
    在金融风控场景中，通过动态修改 LCEL 表达式（如调整风险评分阈值 score > 0.7 → score > 0.75），实时更新风控流程，而无需修改代码。
4. 提升组件复用性与可测试性


  • 组件复用：
    标准化的 LCEL 接口使组件可插拔。例如，替换 LLM 提供商时，只需修改组件实例（如从 OpenAI 改为 Cohere），LCEL 表达式无需变更。
  • 独立测试：
    可单独测试 LCEL 表达式的逻辑正确性，如：
    1. from langchain.lcel import evaluate
    3. # 测试条件表达式
    4. result = evaluate("{{ score > 0.8 ? 'pass' : 'fail' }}", context={"score": 0.9})
    5. assert result == "pass"
    复制代码
    无需启动完整的链或依赖外部服务，提升测试效率。
5. 适配多模态与复杂工具调用

LangChain 常需集成多种工具（如 SQL 查询、API 调用、文件解析），LCEL 可简化多模态数据的处理逻辑：

  • 示例：处理结构化数据
    1. expression = """
    2. let data = sql_tool.query("SELECT * FROM orders WHERE amount > {{ threshold }}");
    3. let summary = llm.summarize(text: json.stringify(data.rows));
    4. return summary;
    5. """
    复制代码

    • 通过 LCEL 链式调用 SQL 工具和 LLM，解耦数据查询与摘要生成逻辑；
    • 表达式清晰描述数据流动路径，避免代码层面的嵌套回调。

三、LCEL 与 LangChain 架构的协同

LCEL 的设计与 LangChain 的模块化架构深度协同，目标是构建 **“表达式驱动的智能应用”**：

  • 底层组件层：提供标准化接口（如 call()、parse()），供 LCEL 表达式调用；
  • LCEL 层：作为逻辑编排层，通过表达式描述组件交互规则；
  • 应用层：通过配置文件、API 动态注入 LCEL 表达式，快速组装不同功能的链。

这种分层设计使 LangChain 既能保持代码的简洁性，又能通过 LCEL 灵活应对复杂业务需求，尤其适合需要频繁调整逻辑的生成式 AI 场景（如智能文档处理、对话系统流程控制等）。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

使用道具 举报
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
兼罔
2025-6-14 14:57:27

0

粉丝关注

18

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫