Runnable 介面

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

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

相關資源

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

Runnable 介面概述

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

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

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

最佳化平行執行（批次）

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

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

兩種批次處理選項是

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

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

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

注意

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

提示

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

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

非同步支援

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

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

串流 API

串流對於使基於 LLM 的應用程式對終端使用者產生快速回應至關重要。

Runnables 公開以下三個串流 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 結構描述，以及輸入和輸出類型的Pydantic 結構描述。

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

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

方法	描述
get_input_schema	提供 Runnable 的輸入結構描述的 Pydantic 結構描述。
get_output_chema	提供 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組合建構的更複雜的 Runnables 效果不佳，並且推斷的輸入和/或輸出類型可能不正確。在這些情況下，我們建議使用者使用 with_types 方法覆寫推斷的輸入和輸出類型（API 參考）。

RunnableConfig

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

RunnableConfig 可以定義以下任何屬性

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

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

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

RunnableConfig 的傳播

許多 Runnables 由其他 Runnables 組成，重要的是 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 傳遞給 Runnable。

這是由於 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

設定遞迴限制

注意

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

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

設定最大並行數

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

提示

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

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

設定 configurable

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，以執行任意邏輯。當您使用 LangChain 運算式語言 (LCEL) 來組合多個 Runnables，並且需要在其中一個步驟中加入自訂處理邏輯時，這特別有用。

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

• RunnableLambda：使用於不需要串流的簡單轉換。
• RunnableGenerator：當需要串流時，用於更複雜的轉換。

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

重要

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

可配置的 Runnables

注意

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

這有助於配置使用 LangChain 運算式語言 (LCEL) 建立的大型「鏈」，並被 LangServe 用於部署的 Runnables。

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

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

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

請參閱如何配置執行時鏈的內部結構指南，以取得更多關於如何配置執行時鏈的內部結構的資訊。