主要内容：

• LCEL
• Runable interface
• Stream
• Stream events

一、LCEL (LangChain Expression Language )介绍

LangChain 表达式语言

LCEL是一种强大的工作流编排工具，可以从基本组件构建复杂任务连(Chain)，并支持如流式处理、并行处理和日志记录等开箱即用的功能。

支持将原型投入生产，无需更改代码，从最简单的“提示 +LLM”链到最复杂的链。以下是一些亮点

• 一流的流式支持
  当您使用LCEL构建链时，将获得可能得最佳时间到第一个标记（直到输出的第一块内容出现所经过的时间）
  对于某些链，意味着我们直接从LLM流式书传输标记到流式输出解析器，将以与LLM提供程序输出原始标记的速率相同的速度获取解析的增量输出块

• 异步支持
  使用LCEL构建的任何链都可以使用同步API（例如，在您的Jupyter笔记本中进行原型设计）及其异步API（例如，在LangServe服务器中）进行调用。 这使得可以在原型和生产中使用相同的代码，具有出色的性能，并且能够在同一服务器中处理许多并发请求。

• 优化的并行执行
  每当LCEL链具有可以并行执行的步骤时（例如，如果从多个检索其中获取文档），LangChain会自动执行，无论是在同步接口还是异步接口中，以获得可能的最小的延迟。

• 重试和回退
  LCEL链的任何部分配置重试和回退。使得链在规模上更可靠的好方法。 目前正在努力为重试/回退添加流式支持，这样您就可以获得额外的可靠性而无需任何延迟成本。

• 访问中间结果
  对于更复杂的链，访问中间步骤的结果通常非常有用，即使在生成最终输出之前。 这可以用于让最终用户知道正在发生的事情，甚至用于调试您的链。 你可以流式传输中间结果，并且在每个LangServe服务器上都可以使用。

• 输入和输出模式
  输入和输出模型为每个LCEL链提供了从链的结果推断出的Pydantic和JSONSchema模式。 这可用于验证输入和输出，并且LangServe的一个组成部分。

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import  ChatOpenAI
llm = ChatOllama( model="deepseek-r1:7b")
prompt = ChatPromptTemplate.from_template("给我讲个关于{input}的笑话")
parser = StrOutputParser() # 加了这段后，会讲Chunk类转化成字符串，也就是获取Chunk类中的content内容

chain = prompt | llm | parser

response = chain.invoke({"input": "孙悟空"})
print(response)

二、Runable Interface

为了尽可能简化创建自定义链的过程，我们实现了一个Runnable协议。 许多LangChain组件都实现了Runnable协议，

包括聊天模型，LLMS、输出解析器、检索器、提示模版等。 此外，还有一些有用的基本组件可用于处理可运行对象。
这是一个标准的接口，可以轻松定义自定义链，并以标准方式调用它们。 标准接口包括：

• stream: 返回响应的数据块
• invoke: 对输入调用链
• batch：对输入列表调用链
  还有相应的异步方法，与asyncio一起使用await 语法以实现并发
• astream：异步返回响应的数据块
• ainvoke：异步调用链
• abatch:
• astream_log: 异步返回中间步骤，以及最终响应
• astream_events: beta 流式传输链中发生的事件

组件	输入类型	输出类型
提示	字典	提示值
聊天模型	单个字符串、聊天信息或提示值	聊天信息
LLM	单个字符串、聊天信息或提示值	字符串
输出解析器	LLM或聊天模型的输出	取决于解析器
检索器	单个字符串	文档列表

所有可运行对象都公开输入和输出模式，以检查输入和输出：

• input_schema: 从可运行对象结构自动生成的输入Pydantic模型
• output_schema：从可运行对象结构自动生成的输出 Pydantic模型

三、Stream

流式允许对于使基于LLM的应用程序对最终用户的响应性至关重要。 重要的LangChain原语，如聊天模型、输出解析器、提示模板、检索器和代理都实现了LangChain Runnable接口。该接口提供了两种通用的流式表达内容方法：

1. 同步stream和异步 astream：流式传输链的最终输出和默认实现
2. 异步 astream_events和异步astream_log：这些方法提供了一种从链中流式传输中间步骤和最终输出的方式。

这些方法旨在以块的形式流式传输最终输出，尽快返回每个块。 只有在程序中的所有步骤都知道如何处理输入流时，才能进行流式传输。

3.1 同步调用

# 流式传输的一个小例子

from langchain_ollama import  ChatOllama
llm = ChatOllama( model="deepseek-r1:7b", )
chunks = []
for chunk in llm.stream("帮我讲个笑话，100字"):
    chunks.append(chunk)
    print(chunk.content, end="|", flush=True)

3.2 异步调用

import asyncio
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_template("给我讲一个关于{topic}的笑话")

parser = StrOutputParser()

chain = prompt | llm | parser

async def async_stream():
    async for chunk in chain.astream({"topic": "小鸟"}):
        print(chunk, end="|", flush=True)

asyncio.run(async_stream())