Runnable 介面

Runnable 介面是使用 LangChain 組件的基礎，它在許多組件中實現，例如語言模型、輸出解析器、檢索器、編譯後的 LangGraph 圖等等。

本指南涵蓋 Runnable 介面的主要概念和方法，這些概念和方法允許開發人員以一致且可預測的方式與各種 LangChain 組件互動。

相關資源

• 「Runnable」介面 API 參考文件提供了 Runnable 介面及其方法的詳細概述。
• 內建 Runnables 的列表可以在 LangChain Core API 參考文件中找到。當使用 LangChain 運算式語言 (LCEL) 在 LangChain 中組合自訂「鏈」時，其中許多 Runnable 非常有用。

Runnable 介面概觀

Runnable 方式定義了一個標準介面，允許 Runnable 組件

• 調用：單個輸入被轉換為輸出。
• 批次處理：多個輸入被有效地轉換為輸出。
• 串流處理：輸出在產生時被串流。
• 檢視：可以存取關於 Runnable 的輸入、輸出和配置的示意資訊。
• 組合：可以使用 LangChain 運算式語言 (LCEL) 組合多個 Runnable 以協同工作，從而建立複雜的管道。

請查看 LCEL 速查表，以了解一些涉及 Runnable 介面和 LCEL 運算式的常見模式。

優化平行執行（批次）

LangChain Runnable 提供內建的 batch（和 batch_as_completed）API，允許您平行處理多個輸入。

當需要處理多個獨立輸入時，使用這些方法可以顯著提高效能，因為可以平行而不是循序地完成處理。

兩個批次處理選項為

• batch：平行處理多個輸入，並以與輸入相同的順序傳回結果。
• batch_as_completed：平行處理多個輸入，並在完成時傳回結果。結果可能會亂序到達，但每個結果都包含用於匹配的輸入索引。

batch 和 batch_as_completed 的預設實作使用線程池執行器平行執行 invoke 方法。這允許高效的平行執行，而無需使用者管理線程，並加速 I/O 綁定程式碼（例如，發出 API 請求、讀取檔案等）。對於 CPU 綁定操作，它不會那麼有效，因為 Python 中的 GIL（全域解釋器鎖）會阻止真正的平行執行。

某些 Runnable 可能提供它們自己的 batch 和 batch_as_completed 實作，這些實作針對其特定用例進行了優化（例如，依賴模型提供者提供的 batch API）。

注意

abatch 和 abatch_as_completed 的非同步版本依賴 asyncio 的 gather 和 as_completed 函數平行執行 ainvoke 方法。

提示

當使用 batch 或 batch_as_completed 處理大量輸入時，使用者可能希望控制最大平行呼叫數。這可以透過在 RunnableConfig 字典中設定 max_concurrency 屬性來完成。請參閱 RunnableConfig 以取得更多資訊。

聊天模型也具有內建的 速率限制器，可用於控制發出請求的速率。

非同步支援

Runnable 公開了非同步 API，允許使用 Python 中的 await 語法調用它們。非同步方法可以通過「a」前綴來識別（例如，ainvoke、abatch、astream、abatch_as_completed）。

請參閱 使用 LangChain 進行非同步程式設計 指南以取得更多詳細資訊。

串流 API

串流對於使基於 LLM 的應用程式對終端使用者做出快速反應至關重要。

Runnable 公開了以下三個串流 API

1. 同步 stream 和非同步 astream：在產生時產生 Runnable 的輸出。
2. 非同步 astream_events：更進階的串流 API，允許串流中間步驟和最終輸出
3. 傳統版非同步 astream_log：傳統的串流 API，用於串流中間步驟和最終輸出

請參閱 串流概念指南，以取得關於如何在 LangChain 中串流的更多詳細資訊。

輸入和輸出類型

每個 Runnable 都以輸入和輸出類型為特徵。這些輸入和輸出類型可以是任何 Python 物件，並由 Runnable 本身定義。

導致 Runnable 執行的 Runnable 方法（例如，invoke、batch、stream、astream_events）與這些輸入和輸出類型一起使用。

• invoke：接受輸入並傳回輸出。
• batch：接受輸入列表並傳回輸出列表。
• stream：接受輸入並傳回產生輸出的產生器。

輸入類型和輸出類型因組件而異

組件	輸入類型	輸出類型
提示	字典	PromptValue
聊天模型	字串、聊天訊息列表或 PromptValue	ChatMessage
LLM	字串、聊天訊息列表或 PromptValue	字串
輸出解析器	LLM 或聊天模型的輸出	取決於解析器
檢索器	字串	文件列表
工具	字串或字典，取決於工具	取決於工具

請參閱個別組件文件，以取得關於輸入和輸出類型以及如何使用它們的更多資訊。

檢視結構描述

注意

這是一個進階功能，對於大多數使用者來說是不必要的。除非您有檢視 Runnable 結構描述的特定需求，否則您可能應該跳過本節。

在更進階的用例中，您可能希望以程式設計方式檢視 Runnable，並確定 Runnable 期望和產生哪些輸入和輸出類型。

Runnable 介面提供了獲取 Runnable 的輸入和輸出類型的 JSON Schema 以及輸入和輸出類型的 Pydantic 結構描述的方法。

這些 API 主要在內部用於單元測試，並由 LangServe 使用，LangServe 使用這些 API 進行輸入驗證和 OpenAPI 文件的產生。

除了輸入和輸出類型之外，某些 Runnable 還設定了額外的執行階段配置選項。有相應的 API 來獲取 Runnable 的配置選項的 Pydantic 結構描述和 JSON Schema。請參閱 可配置的 Runnable 章節以取得更多資訊。

方法	描述
get_input_schema	提供 Runnable 的輸入結構描述的 Pydantic 結構描述。
get_output_schema	提供 Runnable 的輸出結構描述的 Pydantic 結構描述。
config_schema	提供 Runnable 的配置結構描述的 Pydantic 結構描述。
get_input_jsonschema	提供 Runnable 的輸入結構描述的 JSONSchema。
get_output_jsonschema	提供 Runnable 的輸出結構描述的 JSONSchema。
get_config_jsonschema	提供 Runnable 的配置結構描述的 JSONSchema。

With_types

LangChain 將自動嘗試根據可用資訊推斷 Runnable 的輸入和輸出類型。

目前，這種推斷對於使用 LCEL 組合建立的更複雜的 Runnable 無法很好地工作，並且推斷的輸入和/或輸出類型可能不正確。在這些情況下，我們建議使用者使用 with_types 方法覆蓋推斷的輸入和輸出類型（API 參考文件）。

RunnableConfig

用於執行 runnable 的任何方法（例如，invoke、batch、stream、astream_events）都接受稱為 RunnableConfig 的第二個引數（API 參考文件）。此引數是一個字典，其中包含 Runnable 的配置，該配置將在 runnable 執行期間在執行階段使用。

RunnableConfig 可以定義以下任何屬性

屬性	描述
run_name	用於給定 Runnable 的名稱（不繼承）。
run_id	此呼叫的唯一識別碼。子呼叫將獲得它們自己的唯一執行 ID。
tags	此呼叫和任何子呼叫的標籤。
metadata	此呼叫和任何子呼叫的元數據。
callbacks	此呼叫和任何子呼叫的回呼。
max_concurrency	要進行的最大平行呼叫數（例如，由批次使用）。
recursion_limit	呼叫可以遞迴的最大次數（例如，由傳回 Runnable 的 Runnable 使用）
configurable	Runnable 的可配置屬性的執行階段值。

將 config 傳遞給 invoke 方法的方式如下

some_runnable.invoke(
   some_input, 
   config={
      'run_name': 'my_run', 
      'tags': ['tag1', 'tag2'], 
      'metadata': {'key': 'value'}
      
   }
)

RunnableConfig 的傳播

許多 Runnables 由其他 Runnable 組成，重要的是 RunnableConfig 被傳播到 Runnable 發出的所有子呼叫。這允許將執行階段配置值提供給父 Runnable，這些值由所有子呼叫繼承。

如果不是這種情況，則不可能設定和傳播 回呼 或其他配置值，如 tags 和 metadata，它們預期由所有子呼叫繼承。

建立新 Runnables 主要有兩種模式

1. 使用 LangChain 運算式語言 (LCEL) 以宣告方式

   chain = prompt | chat_model | output_parser

2. 使用 自訂 Runnable（例如，RunnableLambda）或使用 @tool 裝飾器

   def foo(input):
       # Note that .invoke() is used directly here
       return bar_runnable.invoke(input)
   foo_runnable = RunnableLambda(foo)

LangChain 將嘗試自動為這兩種模式傳播 RunnableConfig。

為了處理第二種模式，LangChain 依賴 Python 的 contextvars。

在 Python 3.11 及更高版本中，這可以開箱即用，您無需執行任何特殊操作即可將 RunnableConfig 傳播到子呼叫。

在 Python 3.9 和 3.10 中，如果您正在使用非同步程式碼，則需要在調用 Runnable 時手動將 RunnableConfig 傳遞給它。

這是由於 Python 3.9 和 3.10 中 asyncio 的任務 的限制（它們不接受 context 引數）。

手動傳播 RunnableConfig 的方式如下

async def foo(input, config): # <-- Note the config argument
    return await bar_runnable.ainvoke(input, config=config)
    
foo_runnable = RunnableLambda(foo)

注意

當使用 Python 3.10 或更低版本並編寫非同步程式碼時，RunnableConfig 無法自動傳播，您將需要手動執行！當嘗試使用 astream_events 和 astream_log 串流資料時，這是一個常見的陷阱，因為這些方法依賴於正確傳播在 RunnableConfig 內部定義的 回呼。

設定自訂執行名稱、標籤和元數據

可以使用 RunnableConfig 字典的 run_name、tags 和 metadata 屬性為給定 Runnable 設定自訂執行名稱、標籤和元數據的值。

run_name 是一個字串，可用於為執行設定自訂名稱。此名稱將在日誌和其他位置用於識別執行。它不會被子呼叫繼承。

tags 和 metadata 屬性分別是列表和字典，可用於為執行設定自訂標籤和元數據。這些值由子呼叫繼承。

使用這些屬性對於追蹤和偵錯執行非常有用，因為它們將在 LangSmith 中作為追蹤屬性浮出水面，您可以在其上進行篩選和搜尋。

這些屬性也將傳播到 回呼，並將在串流 API（如 astream_events）中顯示為串流中每個事件的一部分。

相關

• 如何使用 LangChain 追蹤

設定執行 ID

注意

這是一個進階功能，對於大多數使用者來說是不必要的。

您可能需要為給定執行設定自訂 run_id，以防您想稍後參考它或將其與其他系統關聯。

對於每次執行，run_id 必須是有效的 UUID 字串且唯一。它用於識別父執行，子類將自動獲得它們自己的唯一執行 ID。

若要設定自訂 run_id，您可以在調用 Runnable 時將其作為鍵值對傳遞到 config 字典中

import uuid

run_id = uuid.uuid4()

some_runnable.invoke(
   some_input, 
   config={
      'run_id': run_id
   }
)

# Do something with the run_id

設定遞迴限制

注意

這是一個進階功能，對於大多數使用者來說是不必要的。

某些 Runnable 可能會傳回其他 Runnable，如果不正確處理，可能會導致無限遞迴。為了防止這種情況，您可以在 RunnableConfig 字典中設定 recursion_limit。這將限制 Runnable 可以遞迴的次數。

設定最大並行數

如果使用 batch 或 batch_as_completed 方法，您可以在 RunnableConfig 字典中設定 max_concurrency 屬性，以控制並行呼叫的最大數量。當您想要限制並行呼叫的數量，以防止伺服器或 API 過載時，這會很有用。

提示

如果您嘗試對聊天模型發出的請求數量進行速率限制，則可以使用內建的速率限制器，而不是設定 max_concurrency，這樣會更有效。

請參閱如何處理速率限制指南，以取得更多資訊。

設定可配置項

configurable 欄位用於傳遞 Runnable 可配置屬性的執行時期值。

它在 LangGraph 中與 LangGraph 持久性 和 記憶體 頻繁地一起使用。

它在 RunnableWithMessageHistory 中用於類似的目的，以指定 session_id / conversation_id 來追蹤對話歷史記錄。

此外，您可以使用它來指定任何自訂配置選項，以傳遞給他們建立的任何可配置的 Runnable。

設定回呼

使用此選項在執行時期為 runnable 配置回呼。回呼將傳遞給 runnable 進行的所有子呼叫。

some_runnable.invoke(
   some_input,
   {
      "callbacks": [
         SomeCallbackHandler(),
         AnotherCallbackHandler(),
      ]
   }
)

請閱讀回呼概念指南，以取得有關如何在 LangChain 中使用回呼的更多資訊。

重要

如果您在非同步環境中使用 Python 3.9 或 3.10，則在某些情況下，您必須手動將 RunnableConfig 傳播到子呼叫。請參閱傳播 RunnableConfig 章節以取得更多資訊。

從函數建立 runnable

您可能需要建立一個自訂的 Runnable，以執行任意邏輯。如果使用 LangChain Expression Language (LCEL) 來組合多個 Runnables，並且您需要在其中一個步驟中新增自訂處理邏輯，則這特別有用。

有兩種方法可以從函數建立自訂的 Runnable

• RunnableLambda：對於不需要串流的簡單轉換，請使用此方法。
• RunnableGenerator：當需要串流進行更複雜的轉換時，請使用此方法。

請參閱如何執行自訂函數指南，以取得有關如何使用 RunnableLambda 和 RunnableGenerator 的更多資訊。

重要

使用者不應嘗試子類別化 Runnables 來建立新的自訂 Runnable。這比僅僅使用 RunnableLambda 或 RunnableGenerator 更複雜且更容易出錯。

可配置的 runnables

注意

這是一個進階功能，對於大多數使用者來說是不必要的。

它有助於使用 LangChain Expression Language (LCEL) 建立的大型「鏈」的配置，並被 LangServe 用於部署的 Runnables。

有時您可能想要實驗，甚至向最終使用者公開，使用 Runnable 執行操作的多種不同方式。這可能涉及調整聊天模型中的溫度等參數，甚至在不同的聊天模型之間切換。

為了簡化此流程，Runnable 介面提供了兩種方法，用於在執行時期建立可配置的 Runnables

• configurable_fields：此方法允許您配置 Runnable 中的特定屬性。例如，聊天模型的 temperature 屬性。
• configurable_alternatives：此方法使您能夠指定在執行時期可以執行的替代 Runnables。例如，您可以指定可以使用的不同聊天模型列表。

請參閱如何配置執行時期鏈內部元件指南，以取得有關如何配置執行時期鏈內部元件的更多資訊。