1. 鏈條是什么LCEL為什么是“流水線”而不是“代碼”做 LangChain 繞不開兩個(gè)概念鏈條基礎(chǔ)Chains和 LangChain Expression Language。我第一次接觸的時(shí)候也很懵官方教程一上來就是prompt | model | parser這三根豎線看著簡單但它的意義遠(yuǎn)不止語法糖。鏈條本質(zhì)上是一條流水線而 LCEL 就是定義這條流水線的“管道語言”。它把提示詞、模型、輸出解析、記憶、檢索這些零散組件用統(tǒng)一的方式串起來讓整個(gè)調(diào)用過程可以被打印、被流式輸出、被異步執(zhí)行、被回退兜底。簡單說LCEL 解決的問題不是“能不能調(diào)用大模型”而是“調(diào)用完之后這套流程能否像工程一樣被維護(hù)”。適合看這篇內(nèi)容的人我默認(rèn)你已經(jīng)有 LangChain 的基本概念知道什么是 Prompt Template、什么是模型調(diào)用但寫復(fù)雜應(yīng)用時(shí)感覺代碼一團(tuán)亂不知道組件之間怎么協(xié)作。接下來我會(huì)從設(shè)計(jì)邏輯講到代碼級實(shí)操把鏈條的運(yùn)行機(jī)制、內(nèi)部協(xié)議、常見坑點(diǎn)都過一遍。這次是進(jìn)階篇不會(huì)停留在“把兩個(gè)函數(shù)連起來”的層面而是會(huì)聊清楚為什么鏈條值得被當(dāng)作一個(gè)獨(dú)立抽象層來用。先給一個(gè)生活化類比。組裝電腦時(shí)主板上有 PCIe 插槽、內(nèi)存插槽、供電接口每個(gè)硬件只需要符合接口規(guī)范就能互相配合。LCEL 就是 LangChain 的“主板規(guī)范”任何實(shí)現(xiàn) Runnable 協(xié)議的對象都能插進(jìn)管道前一端的輸出自動(dòng)成為后一端的輸入。你不需要關(guān)心每塊“硬件”內(nèi)部怎么工作只需要關(guān)心它暴露出來的輸入輸出形狀。這就是流水線思維和傳統(tǒng)過程式代碼最大的區(qū)別數(shù)據(jù)形態(tài)決定鏈路走向而不是嵌套的調(diào)用順序決定一切。2. 三個(gè)最常用原語以及一次最小運(yùn)行2.1 Prompt、模型、解析器是如何咬合在一起的先看最常見的一條鏈它由三個(gè)原語構(gòu)成ChatPromptTemplate、ChatOpenAI、StrOutputParser。from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser prompt ChatPromptTemplate.from_template(用三句話解釋{topic}適合剛接觸這個(gè)概念的新手) model ChatOpenAI(model_namegpt-4o-mini, temperature0.2) parser StrOutputParser() chain prompt | model | parser這段代碼背后的數(shù)據(jù)流是傳入{topic: 什么是Runnable}Prompt 把它格式化成完整消息列表模型接收消息后返回一個(gè)AIMessage對象解析器把其中的字符串內(nèi)容抽出來最終你拿到的是一個(gè)干凈的str。關(guān)鍵在于|這個(gè)運(yùn)算符。在 Python 里它本來是“或”操作但 LangChain 的組件重寫了__or__方法讓兩邊只要實(shí)現(xiàn)了 Runnable 協(xié)議就能被包成一個(gè)RunnableSequence。第一次看到這個(gè)設(shè)計(jì)時(shí)我覺得它有點(diǎn)“魔法”但用久了會(huì)發(fā)現(xiàn)它極其順手——鏈條的拓?fù)浣Y(jié)構(gòu)直接在代碼里可見你不需要額外定義一堆中間變量。2.2 一個(gè)最小可運(yùn)行例子上面那條鏈保存為chain后可以直接調(diào)用result chain.invoke({topic: LangChain LCEL}) print(result)輸出會(huì)是一段自然語言不再有額外的結(jié)構(gòu)體包裹。如果是在腳本里做批量請求可以改用batchresults chain.batch([ {topic: LCEL}, {topic: Runnable}, {topic: OutputParser}, ])batch內(nèi)部會(huì)盡量并發(fā)執(zhí)行比循環(huán)調(diào)用invoke快不少。如果要做流式返回比如前端聊天框一個(gè)字一個(gè)字蹦效果就使用streamfor chunk in chain.stream({topic: LCEL}): print(chunk, end, flushTrue)2.3 invoke、batch、stream 三條入口的區(qū)別這三兄弟的本質(zhì)都是Runnable接口的執(zhí)行方法只是執(zhí)行模式不同。invoke是單次輸入單次輸出適合請求-響應(yīng)場景batch接收一個(gè)輸入列表內(nèi)部用線程池并發(fā)調(diào)度適合離線批量處理stream則把一個(gè)完整輸出拆成多個(gè) chunk 逐個(gè)吐出適合實(shí)時(shí)生成場景。它們共享同一個(gè)鏈路定義所以你在代碼里只需要切換調(diào)用方法不需要改動(dòng)鏈本身的組裝邏輯。提示不要小看這四個(gè)調(diào)用入口。LCEL 最大的工程價(jià)值之一就是它會(huì)自動(dòng)為鏈提供invoke、batch、stream、ainvoke、abatch、astream六種調(diào)用能力。也就是說只要你的組件實(shí)現(xiàn)了 Runnable 協(xié)議整條鏈天然具備同步、異步、單發(fā)、批量、流式五種產(chǎn)品的適配能力。3. 接線并行、分支、回退鏈條的組裝玩法3.1 用 RunnableParallel 做并行數(shù)據(jù)流很多業(yè)務(wù)場景不是一條直線走到底的。比如做一個(gè)問題分析你可能希望同時(shí)讓模型判斷“用戶情緒”、抽取“關(guān)鍵詞”、并生成“回答提綱”三條結(jié)果最后拼到一起。這時(shí)候用RunnableParallel最合適。from langchain_core.runnables import RunnableParallel analyze_emotion prompt_emotion | model | parser extract_keywords prompt_keywords | model | parser outline prompt_outline | model | parser parallel_chain RunnableParallel( emotionanalyze_emotion, keywordsextract_keywords, outlineoutline, ) result parallel_chain.invoke({input: 我很難過想了解怎么緩解焦慮})RunnableParallel接收的每個(gè)關(guān)鍵字參數(shù)都是一條子鏈。執(zhí)行時(shí)這幾條子鏈會(huì)并發(fā)運(yùn)行最后返回一個(gè)字典key 就是參數(shù)名value 就是對應(yīng)子鏈的結(jié)果。有個(gè)細(xì)節(jié)值得注意RunnableParallel的輸入會(huì)原樣復(fù)制給每個(gè)子鏈。也就是說它并不負(fù)責(zé)“分發(fā)不同的輸入”而是“共享同一份輸入”。如果某個(gè)子鏈需要對輸入做不同切割你得在它前面再接一個(gè)RunnableLambda做預(yù)處理。3.2 用 RunnableBranch 做條件分流當(dāng)鏈路走到某個(gè)節(jié)點(diǎn)時(shí)如果需要根據(jù)中間結(jié)果決定后續(xù)走哪條路可以借助RunnableBranch。它有點(diǎn)像編程語言里的 switch 語句只是每條分支都指向一個(gè)可運(yùn)行對象。from langchain_core.runnables import RunnableBranch branch RunnableBranch( (lambda x: len(x[text]) 50, long_text_chain), (lambda x: ? in x[text], question_chain), default_chain, )每個(gè)條件是一個(gè)二元組第一個(gè)元素是判斷函數(shù)返回布爾值第二個(gè)元素是對應(yīng)要執(zhí)行的鏈。執(zhí)行時(shí)按順序逐個(gè)判斷一旦命中就走向?qū)?yīng)分支全部不命中則使用最后的默認(rèn)鏈。實(shí)際項(xiàng)目里我會(huì)把分支條件和配置外置到 JSON 里讓非技術(shù)人員也能調(diào)整路由規(guī)則。但要注意RunnableBranch判斷時(shí)是同步執(zhí)行的不要在判斷函數(shù)里寫耗時(shí)邏輯否則整個(gè)鏈路會(huì)被卡住。3.3 常用組合算子一覽除了并行和分支還有幾個(gè)常用的組合方式需要掌握。RunnablePassthrough就像一個(gè)“直通線”它不做任何處理直接把輸入原樣傳遞給下一步但常用于在并行節(jié)點(diǎn)中把原始輸入保留下來。RunnableLambda可以把任意 Python 函數(shù)包裝成 Runnable用來做輕量級的數(shù)據(jù)清洗、格式轉(zhuǎn)換、校驗(yàn)。from langchain_core.runnables import RunnablePassthrough, RunnableLambda def clean_text(text: str) - str: return text.strip().replace(\n, ) chain ( RunnablePassthrough() | RunnableLambda(lambda x: {question: clean_text(x[question])}) | prompt | model | parser )還有一個(gè)絕不能忽視的.config參數(shù)體系。所有 Runnable 在調(diào)用時(shí)都可以傳入配置對象比如重試次數(shù)、超時(shí)時(shí)間、回調(diào)處理器。組合鏈路會(huì)自動(dòng)向下透傳這些配置保證你在任何一個(gè)節(jié)點(diǎn)上都能夠做到可觀測。4. 鏈內(nèi)部協(xié)議Runnable 接口與數(shù)據(jù)流約定4.1 Runnable 到底規(guī)定了什么要真正理解鏈條必須把Runnable這個(gè)協(xié)議看透。它是 LangChain 定義的一套統(tǒng)一接口規(guī)范任何實(shí)現(xiàn)它的對象都能參與鏈路組裝。核心方法有三個(gè)invoke、batch、stream以及各自對應(yīng)的異步版本。此外還有一個(gè)get_graph()方法用于返回鏈路的圖結(jié)構(gòu)這也是可視化調(diào)試的基礎(chǔ)。鏈條執(zhí)行時(shí)內(nèi)部會(huì)維護(hù)一個(gè)RunnableConfig對象它是貫穿整個(gè)鏈路的“隱形背包”?；卣{(diào)處理器、標(biāo)簽、元數(shù)據(jù)、最大并發(fā)數(shù)、遞歸限制都放在這個(gè)配置里。子鏈執(zhí)行時(shí)不會(huì)重新創(chuàng)建配置而是從入口一層層透傳下去。這也是為什么你可以只在外層調(diào)用chain.invoke卻能在日志里看到每一步的執(zhí)行細(xì)節(jié)。4.2 中間結(jié)果去哪了新手最常問的一個(gè)問題是鏈條中間的每一步結(jié)果我怎么拿默認(rèn)情況下chain.invoke只返回最終輸出中間狀態(tài)不會(huì)暴露。如果你想觀察中間環(huán)節(jié)有幾種做法。第一種是臨時(shí)插入一個(gè)RunnableLambda打印當(dāng)前數(shù)據(jù)def debug_step(x): print(debug:, x) return x chain prompt | RunnableLambda(debug_step) | model | parser第二種是通過回調(diào)函數(shù)切入。LangChain 提供BaseCallbackHandler你可以繼承它并覆寫on_llm_start、on_chain_end等事件在鏈路運(yùn)行過程中接收狀態(tài)通知。第三種更徹底用langchain_core.runnables.utils里的中間結(jié)果追蹤機(jī)制在invoke時(shí)傳入config{callbacks: [...]}把每一步的事件收集起來。我的個(gè)人建議是開發(fā)調(diào)試階段用第一種簡單粗暴生產(chǎn)環(huán)境用第二種把關(guān)鍵事件發(fā)送到日志系統(tǒng)或監(jiān)控面板避免在業(yè)務(wù)代碼里到處插打印語句。4.3 輸入輸出形狀的一致性設(shè)計(jì)LCEL 之所以能串起來背后有個(gè)非常樸素的約定前一節(jié)點(diǎn)的輸出必須能被后一節(jié)點(diǎn)接受。但這個(gè)約定在日常編碼中并不總能滿足最典型的就是模型返回AIMessage對象而下一步解析器期待的是字符串或消息對象。這些包裝對象把格式差異“隱藏”了一旦你試圖把自定義函數(shù)直接插進(jìn)鏈里形狀問題立刻暴露。我在做復(fù)雜鏈路時(shí)會(huì)強(qiáng)制自己先畫一張“數(shù)據(jù)形狀變化表”每一步輸入是什么類型、輸出是什么類型。表格寫在代碼注釋里鏈路一長這張表比任何文檔都管用。下面是一個(gè)例子節(jié)點(diǎn)輸入形狀輸出形狀PromptdictMessage 列表ModelMessage 列表AIMessageParserAIMessagestrRunnableLambdastrdict只要這張表的每一列能對齊鏈路就能跑通。任何一步對不上就是你調(diào)試的第一切入點(diǎn)。5. 實(shí)操給一個(gè)虛構(gòu)知識庫問答系統(tǒng)組裝索引鏈路5.1 需求拆解從文檔切片到檢索結(jié)果為了把前面這些概念落到地面我創(chuàng)建一個(gè)模擬項(xiàng)目X一個(gè)企業(yè)內(nèi)部知識庫問答系統(tǒng)。它的基礎(chǔ)需求是用戶上傳一批文檔系統(tǒng)把文檔拆成小塊為每一塊生成向量索引用戶提問時(shí)檢索相關(guān)塊再交給大模型回答。在 LangChain 語境下這個(gè)流程可以用 LCEL 串起來。先拆數(shù)據(jù)流輸入是一段用戶查詢字符串。把查詢文本轉(zhuǎn)成嵌入向量。這里注意查詢向量和文檔向量的模型必須一致否則維度都對不上。用向量在向量庫中做相似度檢索拿到最相關(guān)的 Top-K 文檔塊。把用戶查詢和檢索到的文檔塊一起塞進(jìn) Prompt。由模型生成回答并帶引用的來源標(biāo)記。聽起來步驟很多但用 LCEL 組織后整體代碼可以控制得很整潔。5.2 用 LCEL 串聯(lián)全流程下面是這個(gè)模擬項(xiàng)目的核心鏈路省略了向量庫連接細(xì)節(jié)只看 LCEL 組織方式from langchain_core.runnables import RunnablePassthrough, RunnableLambda from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_core.prompts import ChatPromptTemplate # 假設(shè) retriever 已經(jīng)是一個(gè) RAG 風(fēng)格的可檢索對象 retriever vector_store.as_retriever(search_kwargs{k: 4}) def format_docs(docs): return \n\n.join(f[來源:{d.metadata.get(source, 未知)}]\n{d.page_content} for d in docs) prompt ChatPromptTemplate.from_template( 請根據(jù)以下資料回答問題。 資料 {context} 問題{question} 要求回答時(shí)標(biāo)注使用了哪些來源。 ) chain ( RunnablePassthrough.assign(contextlambda x: format_docs(retriever.invoke(x[question]))) | prompt | ChatOpenAI(model_namegpt-4o-mini, temperature0) | StrOutputParser() ) result chain.invoke({question: 如何配置開發(fā)環(huán)境的訪問憑證}) print(result)這里的核心技巧是RunnablePassthrough.assign。它接收一個(gè)或多個(gè)匿名函數(shù)把原始輸入字典復(fù)制一份并在其中添加新的 key。這樣原始question字段會(huì)一直保留到 Prompt 階段而檢索得到的文檔塊被格式化成context參與生成。如果你不用assign就得自己維護(hù)一個(gè)狀態(tài)字典鏈路稍長一點(diǎn)就開始漏字段。5.3 現(xiàn)場踩坑記錄與結(jié)果驗(yàn)證這條鏈路我第一次跑通時(shí)遇到三個(gè)典型問題。第一個(gè)問題檢索器返回的文檔對象不能直接塞進(jìn) Prompt。如果不調(diào)用format_docs模型會(huì)收到一串Document對象很多模型序列化后產(chǎn)生大量元數(shù)據(jù)噪聲回答質(zhì)量明顯下降。后來我在代碼里強(qiáng)制要求所有進(jìn)入 Prompt 的內(nèi)容都必須是純字符串結(jié)構(gòu)化數(shù)據(jù)在進(jìn) Prompt 之前統(tǒng)一轉(zhuǎn)好。第二個(gè)問題retriever.invoke是阻塞調(diào)用放在鏈路中間會(huì)讓整體響應(yīng)時(shí)間變長。解決方案是實(shí)現(xiàn)一個(gè)異步檢索包裝器通過RunnableLambda的afunc參數(shù)注冊異步函數(shù)讓檢索和后續(xù)生成鏈路全部走ainvoke。async def aretrieve(question: str) - list: docs await retriever.ainvoke(question) return format_docs(docs) chain ( RunnablePassthrough.assign(contextRunnableLambda(aretrieve, afuncaretrieve)) | prompt | model | parser )第三個(gè)問題是最隱蔽的Temperature 參數(shù)對 RAG 場景的影響。我一開始把模型溫度設(shè)成 0.7結(jié)果同一個(gè)問題連續(xù)問兩次答案措辭差異很大并且有一次模型竟然忽略檢索到的資料直接憑“記憶”回答。把溫度降到 0 之后輸出基本穩(wěn)定并且更愿意引用輸入資料中的原文。6. 常見問題與排查經(jīng)驗(yàn)6.1 流式輸出突然中斷或變亂LCEL 的stream并不保證每一層都支持流式傳遞。比如RunnableParallel內(nèi)部各子鏈并行執(zhí)行輸出聚合時(shí)天然有“等所有結(jié)果齊了才能繼續(xù)”的階段所以流式體驗(yàn)會(huì)打折。另一個(gè)常見坑是StrOutputParser流式工作正常但一旦中間插入RunnableLambda這個(gè)函數(shù)如果是普通同步函數(shù)它會(huì)成為保底策略在收到全部數(shù)據(jù)后才執(zhí)行整個(gè)鏈路就被迫退化為“非流式”。排查思路是先逐段測試直接對model | parser流式調(diào)用確認(rèn)模型層支持 SSE再逐步往前加節(jié)點(diǎn)定位到哪一段把流式能力掐斷了。另外注意檢查 LangChain 版本老版本的某些運(yùn)營商模型不支持流式這是庫層面的限制不是你的代碼問題。6.2 鏈路中某個(gè)節(jié)點(diǎn)報(bào)錯(cuò)重試機(jī)制不生效LCEL 的.with_retry()可以給整個(gè)鏈條加重試策略但它只對可重試的運(yùn)算符生效。如果你自定義的RunnableLambda內(nèi)部拋出異常默認(rèn)情況下重試會(huì)按配置次數(shù)重新執(zhí)行整個(gè) lambda。問題在于 lambda 如果帶有副作用比如寫文件、調(diào)用外部接口重復(fù)執(zhí)行可能造成重復(fù)寫入或重復(fù)扣費(fèi)。我在集成外部搜索 API 時(shí)就踩過這個(gè)坑重試三次等于調(diào)用三次搜索服務(wù)。后來我調(diào)整了做法把有副作用的操作拆出來放到鏈路外面的獨(dú)立服務(wù)中鏈路內(nèi)只保留純計(jì)算邏輯?；蛘呓o搜索調(diào)用單獨(dú)包裝一個(gè)限流和去重邏輯確保相同請求在重試窗口內(nèi)只真正外呼一次。6.3 輸入輸出鍵名不一致鏈路毫無報(bào)錯(cuò)但結(jié)果為空這種情況最折磨人。鏈路跑通了也不報(bào)錯(cuò)但最終輸出是空字符串或者文檔塊沒被帶入上下文。我排查出來的原因通常是某個(gè)節(jié)點(diǎn)重新構(gòu)造了字典丟掉了原始 key 或改變了 key 名。比如有人習(xí)慣在中間寫return {query: ...}但 Prompt 模板期待的是question字段。由于 prompt 對缺失 key 的處理方式在不同版本里不一樣有時(shí)它會(huì)靜默跳過而不是拋異常最后模型拿到空內(nèi)容產(chǎn)出空回答。我的習(xí)慣是所有鏈路入口參數(shù)統(tǒng)一定義成一個(gè)數(shù)據(jù)類或一個(gè)帶 schema 的 TypedDict每個(gè)節(jié)點(diǎn)只修改自己負(fù)責(zé)的字段絕不在局部新建“形狀不同”的字典。6.4 調(diào)試鏈條結(jié)構(gòu)的小技巧chain.get_graph()配合.print_ascii()可以把鏈路結(jié)構(gòu)打印成樹狀圖這個(gè)功能在排查“鏈路到底連了哪些節(jié)點(diǎn)”時(shí)非常好用。舉個(gè)例子如果你寫了一個(gè)鏈條但不確定內(nèi)部是不是組合成了單一序列打印出來一看便知。此外還可以用.get_prompts()在鏈路中提取所有被使用的 Prompt 模板方便做提示詞版本管理。實(shí)際項(xiàng)目里我會(huì)把鏈條結(jié)構(gòu)、數(shù)量、平均耗時(shí)、Token 用量統(tǒng)一打到日志系統(tǒng)。LCEL 本身支持標(biāo)簽和元數(shù)據(jù)透傳在調(diào)用時(shí)加上config{metadata: {chain: knowledge_qa}}即可排查線上問題時(shí)能夠按鏈路維度篩選日志效率高很多。7. 經(jīng)驗(yàn)總結(jié)鏈條不是炫技是工程邊界寫到這里我個(gè)人對 LCEL 最深的體會(huì)是它最大的價(jià)值不是省了幾行代碼而是讓“組件之間的邊界”變得清晰。以前寫大模型應(yīng)用容易把所有邏輯堆在幾個(gè)函數(shù)里開關(guān)參數(shù)滿天飛。現(xiàn)在用鏈條后每個(gè)節(jié)點(diǎn)只做一件事數(shù)據(jù)流一眼就能看穿單元測試也可以針對單個(gè)節(jié)點(diǎn)獨(dú)立寫。生產(chǎn)環(huán)境里掛在鏈上的每一個(gè)環(huán)節(jié)都能單獨(dú)替換不影響上下游新同學(xué)接手代碼時(shí)也不用從第一個(gè)函數(shù)讀到最后一行才能知道項(xiàng)目在干什么。如果你剛開始接觸 LCEL我建議不要急著把所有功能都用上。先從prompt | model | parser這三段式的最小鏈路開始跑通流式然后逐步加入RunnablePassthrough、RunnableParallel感受數(shù)據(jù)分發(fā)模式等對數(shù)據(jù)形狀足夠敏感之后再挑戰(zhàn)RunnableBranch和自定義回調(diào)。鏈條越多越能體會(huì)到結(jié)構(gòu)化組裝比過程式堆代碼更抗折騰。最后再分享一個(gè)小技巧寫完一條鏈之后順手把入口參數(shù)、出口結(jié)果的類型定義注釋在鏈接口處例如# 輸入: {question: str}輸出: str。這個(gè)習(xí)慣在別人接手你的項(xiàng)目時(shí)比寫十行文檔都管用。