LangChain 架构演进与功能扩展：流式事件处理、事件过滤及回调策略

LangChain 架构演进与功能扩展

1. LangChain 的演变

v0.1 与 v0.2 的主要差异

v0.1 是初始版本，包含基本功能。从 0.1 到 0.2 完成的关键特性包括：

流式支持增强：通过事件流 API 提供更好的流式支持。
工具调用标准化：Tools Calling 得到标准化支持。
输出结构接口：标准化的输出结构接口。
@chain 装饰器：更容易创建 RunnableLambdas。
异步支持：在 Python 中对许多核心抽象的更好异步支持。
元数据访问：在 AIMessage 中包含响应元数据，方便访问底层模型的原始输出。
可视化：提供可视化的 runnables 或 langgraph 应用的工具。
互操作性：大多数提供商之间的聊天消息历史记录互操作性。
集成包：超过 20 个 Python 流行集成的合作伙伴包。

LangChain 的未来发展

持续致力于 LangGraph 的开发（向 LangGraph 迁移），增强代理架构的能力。
重新审视 vectorstores 抽象，以提高可用性和可靠性。
改进文档和版本化文档。
计划在 7 月至 9 月之间发布 0.3.0 版本，全面支持 Pydantic 2，并停止对 Pydantic 1 的支持。

注意：自 0.2.0 版本起，langchain 不再依赖 langchain-community。langchain-community 将依赖于 langchain-core 和 langchain。

具体变化

从 0.2.0 版开始，langchain 必须与集成无关。这意味着，langchain 中的代码默认情况下不应实例化任何特定的聊天模型、LLMs、嵌入模型、vectorstores 等；相反，用户需要明确指定这些模型。

以下这些 API 从 0.2 版本起要显式的传递 LLM：

langchain.natbot.NatBotChain.from_default 已被移除，推荐使用 from_llm 类方法。

@tool 修饰符的变化

0.2 前运行结果会显示工具签名：my_tool: (x: str) -> str - Some description. 0.2 后的运行结果简化为：Some description.

LangChain 生态系统定位

LangChain 简化了 LLM 应用程序生命周期的每个阶段：

开发：使用 LangChain 的开源构建块、组件和第三方集成构建您的应用程序。使用 LangGraph 构建具有一流流媒体和人机交互支持的状态代理。
生产化：使用 LangSmith 检查、监控和评估您的链，以便您可以不断优化和自信地部署。
部署：使用 LangServe 将您的 LangChain 链部署为 REST API。

该框架目前将自身定位为覆盖 LLM 应用开发全生命周期的框架。包含开发、部署、工程化三个大方向：

开发阶段：主要是 python 和 javascript 两种语言的 SDK，配合开放的社区组件模板，来便捷的实现跨 LLM 的 APP 开发。
工程化或产品化阶段：主要是以 LangSmith 为代表的产品，集监控、playground、评估等功能于一身。
部署阶段：主要是 LangServe 产品，基于 fastapi 封装的 LLM API 服务器。

核心包说明

langchain-core：主要的 SDK 依赖包，包括基本的抽象结构和 LECL 脚本语言。
langchain-community：第三方集成。
合作伙伴包（例如 langchain-openai、langchain-anthropic 等）：一些集成被进一步拆分成自己的仅依赖于 langchain-core 的轻量级包。
langchain：构成应用程序认知架构的链、代理和检索策略（剥离后只有 Chains、Agents、以及构成应用程序认知结构的检索策略）。
LangGraph：通过将步骤建模为图中的边和节点，使用 LLM 构建强大且有状态的多参与者应用程序。与 LangChain 顺利集成，但可以在没有 LangChain 的情况下使用。
LangServe：将 LangChain 链部署为 REST API。
LangSmith：功能很多包括提示词模板聚合、监控、调试、评测 LLM 等等，部分功能会收费。

2. 如何迁移到 0.2.x 版本

安装 0.2.x 版本的 langchain-core、langchain，并将可能使用的其他软件包升级到最新版本。（例如，langgraph、langchain-community、langchain-openai 等）。
验证代码是否能在新软件包中正常运行（例如，单元测试通过）。
安装最新版本的 langchain-cli，并使用该工具将代码中使用的旧导入替换为新导入。
手动解决所有剩余的弃用警告。
重新运行单元测试。
如果正在使用 astream_events，请查看如何迁移到 astream events v2。

升级依赖包

0.2 版本对依赖包做了较大调整，详细参照官方文档表格。主要变化是解耦了核心功能与社区集成。

使用 langchain-cli 工具

安装该工具：

pip install langchain-cli

注意：该工具并不完美，在迁移前你应该备份好你的代码。使用的时候您需要运行两次迁移脚本，因为每次运行只能应用一次导入替换。

命令示例：

langchain migrate import --all

3. 基于 Runnables 的流式事件支持

大模型在推理时由于要对下一个字的概率进行计算，所以无论多么牛逼的 LLM，在推理的时候或多或少都有一些延迟。而这种延迟在类似 Chat 的场景里，体验非常不好。除了在 LLM 上下功夫外，提升最明显的就是从用户体验着手，采用类似流式输出的方式，加快反馈提升用户体验，让用户感觉快乐很多，这也是为什么 ChatGPT 会采用这种类似打字机效果的原因。

流式在 LangChain 前面版本已经支持不少，在 0.2 版本里，主要是增加了事件支持，方便开发者对流可以有更细致的操作颗粒度。

流的主要接口

我们知道从 0.1 大版本开始，LangChain 就支持所谓的 Runnable 协议，为大部分组件都添加了一些交互接口，其中流的接口有：

同步方式的 stream 以及异步的 astream：他们会以流的方式得到 chain 的最终结果。
异步方式的 astream_event 和 astream_log：这两个都可以获得到流的中间步骤和最终结果。

3.1 直接使用大模型输出流

from langchain_core.messages import HumanMessage, AIMessageChunk
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-3.5-turbo")
for chunk in llm.stream("你好，世界！"):
    print(chunk.content, end="", flush=True)

3.2 Chain 中的流输出

在使用 Chain 时，同样可以获取流式输出。这允许我们在处理过程中逐步消费数据。

3.3 高级使用：在 Chain 中使用流式输出 JSON 结构

很多时候的实际场景是，我们希望接口输出的是一个 JSON 结构，这样在前端应用层面会比较灵活。但是如果是流式输出，很可能因为字符结构没有输出结束会导致 JSON 报错。这种情况可以这样处理：

我们可以利用 output_parser 结合流式生成，确保每一步都是合法的 JSON 片段，或者等待完整结构后再解析。以下是处理国家信息的示例逻辑：

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

# 定义输出解析器
parser = JsonOutputParser()

# 构建 Chain
chain = (
    {"countries": RunnableLambda(lambda x: [])} 
    # ... 后续逻辑根据实际需求填充
)

在实际应用中，我们期望在这种结构下，可以以流式来取到国家名称。这里需要在 JSON 输出后，继续处理流事件，提取特定字段。

3.4 不支持流式的组件处理（检索器）

并不是所有的组件都支持流式输出，比如检索器就不支持。在原生的 LangChain 中，当你给不支持 stream 的组件调用流接口时，一般不会有打字机效果，而是和使用 invoke 效果差不多。而当你使用 LCEL 去调用类似检索器组件的时候，它依然可以搞出来打字机效果，这也是为什么要尽量使用 LCEL 的原因。

原生检索器在这种情况下只会返回最终结果，并没有流的效果。而使用 LCEL 调用后，则可以输出中间的过程。例如使用 FAISS 作为向量存储：

import faiss
from langchain_community.vectorstores import FAISS
from langchain_core.embeddings import Embeddings

# 假设已初始化 embeddings 和 index
# retriever = FAISS(...).as_retriever()

4. v0.2 的核心特性：流中的事件支持

如要使用该特性，你首先要确认自己的 langchain_core 版本等于 0.2。

import langchain_core
print(langchain_core.__version__)
# '0.2.18'

官方给到了一些注意事项：

使用流要尽量使用异步方式编程。
如果你自定义了函数一定要配置 callback。
不使用 LCEL 的话尽量使用 astream 来访问 LLM。

LangChain 将流的过程细化，并在每个阶段给了开发者一个事件钩子，每个阶段都可以获取输出结果。

4.1 在 ChatModel 中使用

from langchain_core.callbacks import CallbackManagerForLLMRun
from langchain_core.language_models.chat_models import BaseChatModel

# 配置 version=v2 参数表明 events 事件流依然是一个 beta API，后面肯定还有更改，所以商业应用要慎重！
# 该参数只在 langchain-core>=0.2.0 作用！

事件流输出示例：

[
  {
    "event": "on_chat_model_start",
    "data": {"input": "hello"},
    "name": "ChatZhipuAI",
    "run_id": "c87b9c20-6dbf-41d3-989a-0b609c0b3fb4"
  },
  {
    "event": "on_chat_model_stream",
    "data": {"chunk": {"content": "Hello"}},
    "run_id": "c87b9c20-6dbf-41d3-989a-0b609c0b3fb4"
  }
]

4.2 在 Chain 中的使用

在 Chain 内部，事件会按照执行顺序触发，包括 start、end、stream 等。这使得我们可以追踪整个 Chain 的执行路径。

5. 事件过滤

结合事件以及配置参数，可以很方便的找出你想要的阶段数据。通过定义名字实现事件的筛选，后续想要使用的块。

可以使用 tags 参数来标记特定的运行步骤，然后在监听事件时过滤出带有特定 tags 的事件。

config = {"tags": ["retrieval"]}
result = chain.invoke(input, config=config)

6. 回调传播

在工具中使用调用可运行项，则需要将回调传播到可运行项；否则，不会生成任何流事件。

当工具内部调用了 Chain 或其他 Runnable 时，必须确保配置正确传递，以保证事件链路不断裂。

{'event': 'on_tool_start', 'data': {'input': 'hello'}, 'name': 'correct_tool'}
{'event': 'on_chain_start', 'data': {'input': 'hello'}, 'parent_ids': ['...']}
{'event': 'on_chain_end', 'data': {'output': 'olleh'}}
{'event': 'on_tool_end', 'data': {'output': 'olleh'}}

7. 最佳实践总结

优先使用异步：在长耗时操作中，务必使用 astream 和 astream_events 以避免阻塞主线程。
明确依赖：在 v0.2+ 中，避免隐式依赖，显式传入 LLM 和 Embedding 模型。
事件监控：利用 on_chain_start/end 和 on_chat_model_stream 构建详细的日志系统，便于调试和成本分析。
JSON 流处理：前端接收流式 JSON 时，建议使用流式解析库（如 eventsource-parser 配合后端流式输出）来处理不完整的 JSON 片段。
迁移测试：升级版本后，务必运行完整的回归测试，特别是涉及工具调用和复杂 Chain 逻辑的部分。

通过上述优化，开发者可以更高效地构建可观测、高性能且易于维护的 LLM 应用。

LangChain 架构演进与功能扩展：流式事件处理、事件过滤及回调策略