为了尽可能简单地创建自定义链，LangChain 提出了“Runnable”标准接口协议。目前，LangChain 中的大多数组件都已实现了 Runnable 协议，用户可以轻松构建自定义链并以标准方式调用它们。其中，标准接口包括：

  • stream：流式输出响应内容块
  • invoke：通过输入来调用链
  • batch：通过批量输入来调用链

对应的异步方法：

  • astream：异步流式调用
  • ainvoke：简单异步调用
  • abatch：批量异步调用
  • astream_log：除了最终响应之外，同时返回中间步骤
  • astream_events：链中发生的流事件（在 langchain-core 0.1.14 中引入）

各组件的输入类型和输出类型列表如下：

组件 输入 输出
Prompt Dictionary PromptValue
ChatModel 单一字符串 / 聊天信息列表 / PromptValue ChatMessage
LLM 单一字符串 / 聊天信息列表 / PromptValue 字符串
OutputParser LLM / ChatModel 的输出 取决于哪个解析器
Retriever 单一字符串 文档列表
Tool 单一字符串或字典，取决于哪个 Tool 取决于哪个 Tool

所有 runnable 组件都暴露输入和输出 schemas 以检查输入和输出：

  • input_schema ：从 Runnable 结构自动生成的输入 Pydantic 模型
  • output_schema ：从 Runnable 结构自动生成的输出 Pydantic 模型

Let’s take a look at these methods. To do so, we’ll create a super simple PromptTemplate + ChatModel chain.
下面，我们将创建一个超级简单的 PromptTemplate + ChatModel 链来看看如何使用这些方法。

1
2
3
4
5
6
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

model = ChatOpenAI()
prompt = ChatPromptTemplate.from_template("tell me a joke about {topic}")
chain = prompt | model

Input Schema

Input Schema 是 Runnable 对象所接收的输入的一个描述，是一个从任何 Runnable 的结构动态生成的 Pydantic 模型，可以调用 .schema() 方法来获取这个 JSON Schema 表示形式。

1
2
3
4
chain.input_schema.schema()
prompt.input_schema.schema()
model.input_schema.schema()

Output Schema

可以调用 .schema() 方法来获取这个 JSON Schema 表示形式

1
2
3
chain.output_schema.schema()
prompt.output_schema.schema()
model.output_schema.schema()

各接口调用方式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 流式处理
for s in chain.stream({"topic": "bears"}):
    print(s.content, end="", flush=True)

# 调用
chain.invoke({"topic": "bears"})

# 批量
chain.batch([{"topic": "bears"}, {"topic": "cats"}])
chain.batch([{"topic": "bears"}, {"topic": "cats"}], config={"max_concurrency": 5}) # 最大并发

# 异步流
async for s in chain.astream({"topic": "bears"}):
    print(s.content, end="", flush=True)

# 异步调用 
await chain.ainvoke({"topic": "bears"})

# 异步批量
await chain.abatch([{"topic": "bears"}])