人工智能大模型AI AgentAgent 框架RAG【免费下载链接】langchainThe agent engineering platform.项目地址https://gitcode.com/GitHub_Trending/la/langchain点击查看免费下载本篇技术指南围绕 LangChain 核心库langchain_core的Runnable 协议组合性Composability展开系统讲解 LangChain Expression LanguageLCEL如何通过|管道符、字典字面量、RunnableBranch、RouterRunnable等原语声明式地串联、并行化与条件路由组件并自动继承同步/异步调用、批处理与流式输出能力。读完本文你将掌握从「提示词 → 模型 → 解析器」经典链路到 Fan-Out/Fan-In 并行处理、Fallback 容错、自定义 Runnable 等全套实战方案并深入理解其底层实现原理。组合性Runnable 协议的核心能力组合性Composability是 LangChain Runnable 协议最核心的特性以声明式方式对组件进行串联、并行化与条件路由。任何由 Runnable 组合而成的链都会自动继承以下四类执行能力同步调用invoke异步调用ainvoke批处理batch/abatch流式输出stream/astream且这些能力天然带有针对效率的优化如批处理复用线程池、流式全链路透传。Runnable 协议的定义与全部方法签名位于 runnables/base.py协议细节可进一步参考 openwiki/runnables.md。组合的两大核心原语分别是RunnableSequence通过|操作符实现的顺序链式组合RunnableParallel通过字典语法实现的并行执行。条件路由则由RunnableBranch谓词路由和RouterRunnable键值路由承担。顺序组合|管道操作符管道操作符|将多个 Runnable 依序串联每一步的输出自动成为下一步的输入。这是最常用的组合模式from langchain_core.runnables import RunnableLambda add_one RunnableLambda(lambda x: x 1) mul_two RunnableLambda(lambda x: x * 2) sequence add_one | mul_two sequence.invoke(1) # (1 1) * 2 4从源码看|操作符由Runnable.__or__实现见 runnables/base.py其核心逻辑就是def __or__(self, other): Runnable or operator. Compose this Runnable with another to create RunnableSequence. return RunnableSequence(self, coerce_to_runnable(other))也就是说|会构造一个RunnableSequence该序列具备以下特性依序调用每个步骤将上一步输出传给下一步输入扁平化嵌套序列避免不必要的嵌套开销详见后文「优化与扁平化」一节若所有步骤都实现了transform将Iterator[Input] → Iterator[Output]自动保持流式属性透传同时支持同步与异步执行。数据流Input → Step 1 → Step 2 → Step 3 → Output当字典被管道进序列时它会被自动转换为RunnableParallelsequence add_one | { mul_2: RunnableLambda(lambda x: x * 2), mul_5: RunnableLambda(lambda x: x * 5), } sequence.invoke(1) # {mul_2: 4, mul_5: 10}除|之外Runnable还提供等价的显式方法形式pipe(*others, nameNone)例如runnable_1.pipe(runnable_2, runnable_3)与runnable_1 | runnable_2 | runnable_3完全等价适合在需要动态构造链的场景下使用。并行组合字典语法与 RunnableParallel并行执行指多个 Runnable 在同一份输入上并发运行。在序列中通过字典字面量即可实现也可以直接显式使用RunnableParallel。字典字面量语法from langchain_core.runnables import RunnableLambda, RunnableParallel add_one RunnableLambda(lambda x: x 1) mul_two RunnableLambda(lambda x: x * 2) mul_three RunnableLambda(lambda x: x * 3) # Dict syntax creates a RunnableParallel sequence add_one | { mul_2: mul_two, mul_3: mul_three, } sequence.invoke(1) # Output: {mul_2: 4, mul_3: 6}显式 RunnableParallelparallel RunnableParallel( mul_2mul_two, mul_3mul_three, ) parallel.invoke(2) # Output: {mul_2: 4, mul_3: 6}并发执行机制从 runnables/base.py 中RunnableParallel的实现可以看出其并发细节输入副本隔离RunnableParallel使用atee异步或safetee同步为每个分支创建独立的输入副本避免分支间互相污染并发调度同步执行走线程池异步执行走asyncio.gather配合gather_with_concurrency进行并发度控制见 runnables/utils.py异步流式使用asyncio.wait(return_whenFIRST_COMPLETED)管理任务哪个分支先产出 chunk 就先发射哪个实现低延迟输出结果合并最终以字典形式汇总所有分支的输出内部使用AddableDict累加各分支结果。批处理对多个输入并行调用批处理Batching让一条管道高效地处理多个输入。与并行分支同一输入跑多条链不同批处理是同一条序列分别应用于每个输入同步批处理sequence add_one | mul_two results sequence.batch([1, 2, 3]) # [4, 6, 8] # Each input processed in parallel via thread pool异步批处理results await sequence.abatch([1, 2, 3]) # [4, 6, 8]底层实现默认的batch通过get_executor_for_config获取线程池执行器见 runnables/config.py对每个输入并行调用invokeabatch使用asyncio.gather并通过配置项max_concurrency限制并发度序列内每个步骤独立批处理其输入RunnableSequence依序对每个步骤调用batch把上一步的批结果喂给下一步。这意味着只要链中某一步如 LLM 的批 API实现了批处理优化整条链都能获益此外还提供batch_as_completed/abatch_as_completed按完成顺序产出(索引, 结果)元组适合需要边处理边消费部分结果的场景。流式输出逐 Token 返回流式Streaming在输出产生的同时即时发射 chunk让 LLM 等顺序生成器实现实时响应for chunk in sequence.stream(1): print(chunk) # Intermediate outputs as they become available异步版本async for chunk in sequence.astream(1): print(chunk) # Non-blocking iteration流式管道透传规则RunnableSequence对流式属性有精密的保持策略若所有步骤都实现transform即把输入迭代器转换为输出迭代器流式将全链路透传若任一环节不支持transform流式会在该步骤处阻塞等该步骤完成后恢复后续发射RunnableLambda默认不实现transform需要自定义流式逻辑时请使用RunnableGenerator它直接接收Iterator[Input]并产出Iterator[Output]。实战示例Prompt → Model → Parser 全链路流式from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser prompt ChatPromptTemplate.from_template(What is {topic}?) model ChatOpenAI() parser StrOutputParser() chain prompt | model | parser # Stream tokens as the model generates them for chunk in chain.stream({topic: composability}): print(chunk, end, flushTrue)该链中的数据流转ChatPromptTemplate将输入字典格式化为字符串提示词ChatOpenAI边从 API 接收边流式发射 tokenStrOutputParser将 token 原样透传。Token 端到端流动无需等待完整响应即可开始消费。条件路由RunnableBranch 与 RouterRunnable条件逻辑根据谓词或键值将输入路由到不同分支。RunnableBranch基于谓词的路由RunnableBranch依序求值各条件执行第一个命中条件的对应分支from langchain_core.runnables import RunnableBranch, RunnableLambda branch RunnableBranch( (lambda x: isinstance(x, int), RunnableLambda(lambda x: x * 2)), (lambda x: isinstance(x, str), RunnableLambda(lambda x: x.upper())), RunnableLambda(lambda x: unknown), ) branch.invoke(5) # 10 branch.invoke(hello) # HELLO branch.invoke(None) # unknown条件依序求值第一个求值为真的条件选中其对应 Runnable若所有条件均不匹配则执行默认分支。从 runnables/branch.py 的源码可确认若干实现约束至少需要两个分支RunnableBranch requires at least two branches_MIN_BRANCHES 2每个分支必须是长度为 2 的(condition, runnable)元组/列表否则抛出TypeError或ValueError默认分支必须是Runnable、可调用对象或Mapping条件和分支 Runnable 都会通过coerce_to_runnable统一转换为 Runnable求值过程会为每个条件/分支创建独立的子回调condition:{idx 1}/branch:{idx 1}标签便于追踪调试。RouterRunnable基于键值的路由RouterRunnable根据输入中的字符串键选择目标 Runnable见 runnables/router.pyfrom langchain_core.runnables.router import RouterRunnable add RunnableLambda(lambda x: x 1) square RunnableLambda(lambda x: x ** 2) router RouterRunnable(runnables{add: add, square: square}) router.invoke({key: square, input: 3}) # 9 router.invoke({key: add, input: 3}) # 4输入是一个包含两个字段的字典key选择路由到哪个 Runnableinput传给被选中 Runnable 的实际数据。源码确认若key不在runnables映射中会抛出ValueError(No runnable associated with key ...)runnables映射中的每个值同样会被coerce_to_runnable规范化。RouterRunnable非常适合调度表dispatch table与多专家multi-expert架构。使用 RunnablePassthrough 保留上下文RunnablePassthrough原样转发输入或附带额外键常用于在并行分支中保留中间上下文实现见 runnables/passthrough.pyfrom langchain_core.runnables import RunnablePassthrough chain ( RunnableLambda(lambda x: x 1) | { original: RunnablePassthrough(), modified: RunnableLambda(lambda x: x * 2), } ) chain.invoke(5) # {original: 6, modified: 12}这里RunnablePassthrough保留了上一步的中间结果 6供另一个分支复用而modified分支则在其上做二次变换。与之配套的还有RunnablePassthrough.assign(**kwargs)生成RunnableAssign在原字典上增量添加键、RunnablePassthrough.pick(keys)RunnablePick从输入字典中选取指定键它们共同构成了字典数据流的增删改查工具集。异步等价方法每个同步方法都有对应的异步版本同步异步invoke(input)ainvoke(input)batch(inputs)abatch(inputs)stream(input)astream(input)transform(Iterator[Input])atransform(AsyncIterator[Input])异步方法深度集成回调系统并通过asyncio.gather执行并发感知的批处理。默认情况下异步方法会经由执行器委托给同步实现支持原生异步的后端如异步 API 客户端会覆写这些方法获得真正的事件循环内执行。变量绑定与上下文流转在组合链中数据与执行上下文execution context一起流经各个步骤每一步接收上一步的输出作为输入。上下文传播Context Propagation链被调用时RunnableSequence会为追踪建立回调层级每个步骤通过run_manager.get_child(fseq:step:{i 1})标记为子 run回调callbacks、标签tags与元数据metadata经RunnableConfig贯穿整条链patch_config在为每个步骤更新配置的同时保留父级上下文见 runnables/config.py。from langchain_core.runnables import RunnableLambda # Context flows through each step step1 RunnableLambda(lambda x: x 1) step2 RunnableLambda(lambda x: x * 2) chain step1 | step2 # Invoke with tracing config result chain.invoke( 5, config{ run_name: my_chain, callbacks: [my_tracer], tags: [prod], } ) # Each step runs with inherited config while reporting to callbacks字典组合与键选择在序列中使用字典语法时每个字典键成为一条独立的分支上下文chain step1 | { result_a: step2, result_b: step3, } # Output combines results from both branches output chain.invoke(input) # {result_a: ..., result_b: ...}每个分支result_a、result_b在回调追踪中都会作为独立的子 run 出现这为性能分析和调试提供了清晰的粒度划分。Fallback 容错模式Fallback 提供容错韧性当某个 Runnable 失败时依次尝试备选 Runnable实现见 runnables/fallbacks.py。组件级 Fallbackfrom langchain_core.runnables import RunnableLambda primary_llm ChatOpenAI(modelgpt-4) fallback_llm ChatAnthropic(modelclaude-3-sonnet) resilient_llm primary_llm.with_fallbacks( [fallback_llm], exceptions_to_handle(APIConnectionError,), ) output resilient_llm.invoke(What is composability?) # Uses primary_llm; falls back to fallback_llm if APIConnectionError occurs链级 Fallback# Construct a chain with fallback chain_with_fallback ( prompt | resilient_llm | parser ).with_fallbacks([ RunnableLambda(lambda x: Service unavailable) ]) output chain_with_fallback.invoke({topic: composability}) # If the entire chain fails, returns fallback response多重 FallbackFallback 依序尝试直到某个成功为止model ChatOpenAI().with_fallbacks([ ChatAnthropic(), # Try second ChatClaude(), # Try third ChatCohere(), # Try fourth RunnableLambda(default_response), # Final fallback ])源码层面的关键行为RunnableWithFallbacks.invoke见 runnables/fallbacks.py默认exceptions_to_handle为(Exception,)即捕获所有标准异常可通过参数收窄为特定异常类型非目标异常会被立即重新抛出逐个执行主 Runnable 与 fallbacks第一个成功即返回全部失败后抛出第一个记录的异常first_error可选参数exception_key若指定会将捕获到的异常以该键注入输入字典传给下一个 fallback此时输入必须是字典否则抛出ValueError每次尝试都会进入回调追踪run_manager.get_child()便于在追踪系统中观察降级过程。典型链式模式通用模式Prompt → Model → Parserfrom langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser chain ( ChatPromptTemplate.from_template(What is {topic}?) | ChatOpenAI() | StrOutputParser() ) # Single invoke output chain.invoke({topic: LLMs}) # Batch process outputs chain.batch([{topic: LLMs}, {topic: Vectors}]) # Stream tokens for chunk in chain.stream({topic: LLMs}): print(chunk, end, flushTrue)同一条链无需任何额外代码即可同时获得单次调用、批处理与流式三种执行模式——这正是 Runnable 协议组合性的价值所在。Fan-Out / Fan-In并行处理from langchain_core.runnables import RunnableLambda, RunnablePassthrough chain ( RunnablePassthrough() | { summary: RunnableLambda(summarize), entities: RunnableLambda(extract_entities), sentiment: RunnableLambda(analyze_sentiment), } ) result chain.invoke(text) # {summary: ..., entities: [...], sentiment: positive}条件执行from langchain_core.runnables import RunnableBranch route_logic RunnableBranch( (lambda x: math in x.lower(), math_chain), (lambda x: code in x.lower(), code_chain), general_chain, ) output route_logic.invoke(How do I calculate factorial?)类型安全与 Schema 推断链会从各组件自动推断输入/输出类型sequence add_one | mul_two # Access inferred schemas print(sequence.input_schema) # Pydantic model for input print(sequence.output_schema) # Pydantic model for output print(sequence.input_schema.model_json_schema())schema是从泛型类型参数或实现者提供的类型标注推断出来的 Pydantic 模型见 runnables/base.py 中的get_input_schema/get_output_schema/config_schema。这意味着无需显式类型注解即可获得输入输出校验可直接转换为JSON Schema用于 API 文档、外部校验与工具Tool注册as_tool()即依赖此机制自动生成参数 Schema可通过with_types(input_type..., output_type...)手动覆盖推断结果。优化与扁平化RunnableSequence会自动扁平化嵌套序列# These are equivalent: chain1 step1 | step2 | step3 chain2 step1 | (step2 | step3) chain3 (step1 | step2) | step3三者最终都产生一条单一扁平序列[step1, step2, step3]从而避免不必要的嵌套开销如重复的配置合并与回调层级。RunnableSequence内部通过first、middle、last三个字段存储步骤steps()属性可获取扁平化后的步骤列表。序列化与调试组合链通过 LangChain 序列化系统支持持久化Persistence链可保存与加载RunnableSequence、RunnableBranch、RunnableWithFallbacks、RouterRunnable等均为is_lc_serializable追踪Tracing自动集成回调便于通过 LangSmith 等观测工具调试检视Inspection使用get_graph()可视化链结构支持 ASCII 图、Mermaid 与 PNG 输出见 runnables/graph.py。启用调试输出from langchain_core.globals import set_debug set_debug(True) # Print intermediate results chain.invoke(input) # Or use callbacks: from langchain_core.tracers import ConsoleCallbackHandler chain.invoke(input, config{callbacks: [ConsoleCallbackHandler()]})扩展自定义 Runnable实现Runnable接口即可创建自定义组件并立即获得全套组合操作符的兼容性from langchain_core.runnables import Runnable, RunnableConfig from typing import Iterator class CustomRunnable(Runnable[str, int]): def invoke(self, input: str, config: RunnableConfig | None None) - int: return len(input) async def ainvoke(self, input: str, config: RunnableConfig | None None) - int: return len(input) def stream(self, input: str, config: RunnableConfig | None None) - Iterator[int]: # For streaming support, implement transform for char in input: yield 1 async def astream(self, input: str, config: RunnableConfig | None None): for char in input: yield 1 # Immediately composable chain CustomRunnable() | another_step根据 Runnable 协议见 openwiki/runnables.md子类必须实现抽象方法invoke其余方法batch、stream、异步变体等都有基于invoke的默认实现可按需覆写以利用后端特性如原生异步、批量 API、流式生成。自定义 Runnable 与所有组合操作符天然兼容因此可以无缝融入|序列、字典并行分支、RunnableBranch与 Fallback 体系。组合操作符速查表操作符/方法作用示例\|顺序链式组合step1 \| step2序列中的字典并行分支step1 \| {key1: step2, key2: step3}RunnableBranch谓词条件路由RunnableBranch((cond, runnable), default)RouterRunnable键值路由RouterRunnable({key: runnable}).batch()/.abatch()多输入并行处理chain.batch([in1, in2]).stream()/.astream()逐 Token 输出for chunk in chain.stream(input):.with_fallbacks()容错降级model.with_fallbacks([alt1, alt2]).pipe()显式顺序组合r1.pipe(r2, r3).assign()/.pick()字典增键/选键passthrough.assign(x...)总结LangChain 的组合性建立在统一的 Runnable 协议之上顺序用|并行用字典路由用RunnableBranch/RouterRunnable容错用with_fallbacks保留上下文用RunnablePassthrough。无论组合多么复杂最终产物仍是一个 Runnable因此整条链依旧具备invoke/ainvoke/batch/abatch/stream/astream全套能力且可继续被进一步组合、序列化、追踪与检视。理解这些组合原语及其在 runnables/base.py、runnables/branch.py、runnables/router.py、runnables/fallbacks.py 中的实现细节是构建可维护、可观测、高可用的 LLM 应用链路的基石。Runnable 协议的完整方法签名与默认行为可继续查阅 openwiki/runnables.md。赞分享人工智能大模型AI AgentAgent 框架RAG【免费下载链接】langchainThe agent engineering platform.项目地址https://gitcode.com/GitHub_Trending/la/langchain点击查看免费下载相关推荐LangChain Runnable 核心组合层完全指南LCEL 协议、组合算子与执行模式深入解析LangChain Runnable 核心组合层完全指南LCEL 协议、组合算子与执行模式深入解析 本篇技术指南以 LangChain 开源仓库本仓库 li人工智能大模型AI AgentAgent 框架RAGLangChain LCEL 链式调用原理与实战ai-agents-from-zero 详解 Runnable 与并行链LangChain LCEL 链式调用原理与实战ai agents from zero 详解 Runnable 与并行链 本文为 LangChain LCEL教程人工智能大模型AI AgentRAG提示工程vee-validate 实战指南用 Vue 3 组合式 API 与声明式组件构建无痛表单验证vee validate 实战指南用 Vue 3 组合式 API 与声明式组件构建无痛表单验证 vee validate 是 Vue.js 生态中面向表单验证前端UI组件上一篇如何用Jasminum插件轻松管理中文文献终极Zotero增强指南下一篇KKManager终极游戏模组管理解决方案一键解决14款游戏插件冲突创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
