1. 一個(gè)組件要具備哪些行為最簡(jiǎn)模型是輸入Input→ 處理 → 輸出Output。如果每種組件都各自定義同步、異步、批量、流式的方法名組合起來會(huì)很困難。Runnable 對(duì)這些執(zhí)行方式提供統(tǒng)一詞匯操作用途invoke(x)處理一份輸入ainvoke(x)在 async 程序中處理一份輸入batch(xs)對(duì)多份輸入進(jìn)行批量/并發(fā)處理stream(x)隨結(jié)果產(chǎn)生逐步返回輸出并非所有 Runnable 都原生支持高性能 async、批處理和流式。Base 類有通用實(shí)現(xiàn)具體實(shí)現(xiàn)可按后端特性覆蓋優(yōu)化。2. 從 Python 函數(shù)變成 Runnablefromlangchain_core.runnablesimportRunnableLambdadefword_count(text:str)-int:returnlen(text.split())counterRunnableLambda(word_count)print(counter.invoke(LangChain components can compose))# 4print(counter.batch([one two,three,four five six]))# [2, 1, 3]print(list(counter.stream(one two three)))# [3]RunnableLambda適合把普通函數(shù)放進(jìn)鏈路。此處的 stream 默認(rèn)是一次性產(chǎn)出完整結(jié)果并不意味著函數(shù)本身可以逐 token 生成。3. 通用包裝給現(xiàn)有組件添加策略fromlangchain_core.runnablesimportRunnableLambda runnableRunnableLambda(lambdavalue:value*2)configuredrunnable.with_config(tags[lesson-04],metadata{feature:demo})print(configured.invoke(21))# 42schemarunnable.get_input_jsonschema()print(schema)config里的 tags 和 metadata 可供執(zhí)行上下文與觀測(cè)系統(tǒng)使用。Runnable 還支持.with_retry(...)、.with_fallbacks(...)、.with_listeners(...)等統(tǒng)一包裝而不必讓每個(gè)原始函數(shù)重新實(shí)現(xiàn)。4. 異步調(diào)用示例importasynciofromlangchain_core.runnablesimportRunnableLambda runnableRunnableLambda(lambdavalue:value1)asyncdefmain()-None:resultawaitrunnable.ainvoke(41)print(result)# 42asyncio.run(main())這個(gè)示例可在 Python 腳本中運(yùn)行無需網(wǎng)絡(luò)。默認(rèn)ainvoke可將同步invoke放進(jìn) executorprovider-native async 實(shí)現(xiàn)可以覆蓋它以避免阻塞線程池。5. 源碼設(shè)計(jì)解析統(tǒng)一接口背后的默認(rèn)實(shí)現(xiàn)Runnable是抽象基類不是一種會(huì)自動(dòng)完成業(yè)務(wù)邏輯的對(duì)象。它定義單輸入到單輸出的核心操作invoke組件繼承這一接口就能進(jìn)入 LangChain 的組合與執(zhí)行體系。比如RunnableLambda(word_count)包裝器負(fù)責(zé)接收輸入、調(diào)用 Python 函數(shù)并把結(jié)果作為輸出返回。abstractmethoddefinvoke(self,input,configNone,**kwargs):Transform a single input into an output.asyncdefainvoke(self,input,configNone,**kwargs):returnawaitrun_in_executor(config,self.invoke,input,config,**kwargs)這段基類邏輯說明同步和異步之間的關(guān)系invoke是最小必需的同步契約如果子類沒有高效的原生異步實(shí)現(xiàn)基類會(huì)把同步調(diào)用交給 executor 執(zhí)行避免直接阻塞當(dāng)前 async 事件循環(huán)。它提供的是通用兼容層并不意味著原函數(shù)變成了真正的異步 I/O。需要原生異步的 provider 可以覆寫ainvoke或其底層生成邏輯。同理基礎(chǔ)batch會(huì)在默認(rèn)線程池中并發(fā)執(zhí)行多次invoke并非自動(dòng)調(diào)用某個(gè)供應(yīng)商的批處理 API。支持真正批量請(qǐng)求的子類可以覆寫batch把多條輸入合并為后端的一次請(qǐng)求減少網(wǎng)絡(luò)開銷。統(tǒng)一方法帶來可替換性具體性能仍由實(shí)現(xiàn)決定。通用配置、重試和 fallback 則采用“返回新 Runnable”的裝飾思路而非修改原組件。例如.with_config(tags[...])會(huì)創(chuàng)建RunnableBinding將配置與原 Runnable 綁定.with_retry(...)會(huì)創(chuàng)建RunnableRetry負(fù)責(zé)失敗后的重試策略。于是原本的函數(shù)只負(fù)責(zé)計(jì)算重試、追蹤元數(shù)據(jù)等橫切策略由可組合的包裝器管理。設(shè)計(jì)收益與邊界只實(shí)現(xiàn)一個(gè)清楚的核心行為組件就能獲得一致入口通用默認(rèn)實(shí)現(xiàn)減少接入工作高性能或供應(yīng)商原生語義則保留給具體子類。另一方面batch、stream和ainvoke的存在不等于每個(gè)實(shí)現(xiàn)都以原生并行、細(xì)粒度 token 流或非阻塞 I/O 的方式運(yùn)行。源碼位置以下路徑相對(duì) LangChain 倉庫根目錄libs/core/langchain_core/runnables/base.pyRunnable定義統(tǒng)一執(zhí)行契約及ainvoke、batch等默認(rèn)行為RunnableLambda將 Python callable 包裝為 RunnableRunnableSequence負(fù)責(zé)順序執(zhí)行各子步驟。6. 邊界與錯(cuò)誤認(rèn)識(shí).batch不等價(jià)于供應(yīng)商批量 API。默認(rèn)行為可能是并發(fā)多次 invoke。.stream有統(tǒng)一入口不保證組件每個(gè)都能產(chǎn)生細(xì)粒度 chunks。配置可沿組合流程傳遞但不應(yīng)把敏感憑證寫進(jìn) tags 或 metadata。Runnable 的類型提示和 JSON Schema 幫助表達(dá)契約但動(dòng)態(tài) Python 類型檢查并不會(huì)因此自動(dòng)覆蓋所有運(yùn)行時(shí)錯(cuò)誤。小結(jié)Runnable 是 LangChain 框架“可以組合”的基礎(chǔ)契約。它把單次執(zhí)行、批處理、異步、流式和內(nèi)省放到相似接口下子類可以繼承默認(rèn)行為也能按后端能力優(yōu)化。本系列導(dǎo)航上一篇03 BaseChatModel 與 Provider 適配器——模型如何接入 LangChain下一篇05 LCEL 與 RunnableSequence、RunnableParallel 組合機(jī)制全系列LangChain 源碼入門