1. 一个组件要具备哪些行为最简模型是输入Input→ 处理 → 输出Output。如果每种组件都各自定义同步、异步、批量、流式的方法名组合起来会很困难。Runnable 对这些执行方式提供统一词汇操作用途invoke(x)处理一份输入ainvoke(x)在 async 程序中处理一份输入batch(xs)对多份输入进行批量/并发处理stream(x)随结果产生逐步返回输出并非所有 Runnable 都原生支持高性能 async、批处理和流式。Base 类有通用实现具体实现可按后端特性覆盖优化。2. 从 Python 函数变成 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适合把普通函数放进链路。此处的 stream 默认是一次性产出完整结果并不意味着函数本身可以逐 token 生成。3. 通用包装给现有组件添加策略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 可供执行上下文与观测系统使用。Runnable 还支持.with_retry(...)、.with_fallbacks(...)、.with_listeners(...)等统一包装而不必让每个原始函数重新实现。4. 异步调用示例importasynciofromlangchain_core.runnablesimportRunnableLambda runnableRunnableLambda(lambdavalue:value1)asyncdefmain()-None:resultawaitrunnable.ainvoke(41)print(result)# 42asyncio.run(main())这个示例可在 Python 脚本中运行无需网络。默认ainvoke可将同步invoke放进 executorprovider-native async 实现可以覆盖它以避免阻塞线程池。5. 源码设计解析统一接口背后的默认实现Runnable是抽象基类不是一种会自动完成业务逻辑的对象。它定义单输入到单输出的核心操作invoke组件继承这一接口就能进入 LangChain 的组合与执行体系。比如RunnableLambda(word_count)包装器负责接收输入、调用 Python 函数并把结果作为输出返回。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)这段基类逻辑说明同步和异步之间的关系invoke是最小必需的同步契约如果子类没有高效的原生异步实现基类会把同步调用交给 executor 执行避免直接阻塞当前 async 事件循环。它提供的是通用兼容层并不意味着原函数变成了真正的异步 I/O。需要原生异步的 provider 可以覆写ainvoke或其底层生成逻辑。同理基础batch会在默认线程池中并发执行多次invoke并非自动调用某个供应商的批处理 API。支持真正批量请求的子类可以覆写batch把多条输入合并为后端的一次请求减少网络开销。统一方法带来可替换性具体性能仍由实现决定。通用配置、重试和 fallback 则采用“返回新 Runnable”的装饰思路而非修改原组件。例如.with_config(tags[...])会创建RunnableBinding将配置与原 Runnable 绑定.with_retry(...)会创建RunnableRetry负责失败后的重试策略。于是原本的函数只负责计算重试、追踪元数据等横切策略由可组合的包装器管理。设计收益与边界只实现一个清楚的核心行为组件就能获得一致入口通用默认实现减少接入工作高性能或供应商原生语义则保留给具体子类。另一方面batch、stream和ainvoke的存在不等于每个实现都以原生并行、细粒度 token 流或非阻塞 I/O 的方式运行。源码位置以下路径相对 LangChain 仓库根目录libs/core/langchain_core/runnables/base.pyRunnable定义统一执行契约及ainvoke、batch等默认行为RunnableLambda将 Python callable 包装为 RunnableRunnableSequence负责顺序执行各子步骤。6. 边界与错误认识.batch不等价于供应商批量 API。默认行为可能是并发多次 invoke。.stream有统一入口不保证组件每个都能产生细粒度 chunks。配置可沿组合流程传递但不应把敏感凭证写进 tags 或 metadata。Runnable 的类型提示和 JSON Schema 帮助表达契约但动态 Python 类型检查并不会因此自动覆盖所有运行时错误。小结Runnable 是 LangChain 框架“可以组合”的基础契约。它把单次执行、批处理、异步、流式和内省放到相似接口下子类可以继承默认行为也能按后端能力优化。本系列导航上一篇03 BaseChatModel 与 Provider 适配器——模型如何接入 LangChain下一篇05 LCEL 与 RunnableSequence、RunnableParallel 组合机制全系列LangChain 源码入门 LangChain 源码入门 04：Runnable 如何统一调用、批处理、异步与流式接口
1. 一个组件要具备哪些行为最简模型是输入Input→ 处理 → 输出Output。如果每种组件都各自定义同步、异步、批量、流式的方法名组合起来会很困难。Runnable 对这些执行方式提供统一词汇操作用途invoke(x)处理一份输入ainvoke(x)在 async 程序中处理一份输入batch(xs)对多份输入进行批量/并发处理stream(x)随结果产生逐步返回输出并非所有 Runnable 都原生支持高性能 async、批处理和流式。Base 类有通用实现具体实现可按后端特性覆盖优化。2. 从 Python 函数变成 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适合把普通函数放进链路。此处的 stream 默认是一次性产出完整结果并不意味着函数本身可以逐 token 生成。3. 通用包装给现有组件添加策略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 可供执行上下文与观测系统使用。Runnable 还支持.with_retry(...)、.with_fallbacks(...)、.with_listeners(...)等统一包装而不必让每个原始函数重新实现。4. 异步调用示例importasynciofromlangchain_core.runnablesimportRunnableLambda runnableRunnableLambda(lambdavalue:value1)asyncdefmain()-None:resultawaitrunnable.ainvoke(41)print(result)# 42asyncio.run(main())这个示例可在 Python 脚本中运行无需网络。默认ainvoke可将同步invoke放进 executorprovider-native async 实现可以覆盖它以避免阻塞线程池。5. 源码设计解析统一接口背后的默认实现Runnable是抽象基类不是一种会自动完成业务逻辑的对象。它定义单输入到单输出的核心操作invoke组件继承这一接口就能进入 LangChain 的组合与执行体系。比如RunnableLambda(word_count)包装器负责接收输入、调用 Python 函数并把结果作为输出返回。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)这段基类逻辑说明同步和异步之间的关系invoke是最小必需的同步契约如果子类没有高效的原生异步实现基类会把同步调用交给 executor 执行避免直接阻塞当前 async 事件循环。它提供的是通用兼容层并不意味着原函数变成了真正的异步 I/O。需要原生异步的 provider 可以覆写ainvoke或其底层生成逻辑。同理基础batch会在默认线程池中并发执行多次invoke并非自动调用某个供应商的批处理 API。支持真正批量请求的子类可以覆写batch把多条输入合并为后端的一次请求减少网络开销。统一方法带来可替换性具体性能仍由实现决定。通用配置、重试和 fallback 则采用“返回新 Runnable”的装饰思路而非修改原组件。例如.with_config(tags[...])会创建RunnableBinding将配置与原 Runnable 绑定.with_retry(...)会创建RunnableRetry负责失败后的重试策略。于是原本的函数只负责计算重试、追踪元数据等横切策略由可组合的包装器管理。设计收益与边界只实现一个清楚的核心行为组件就能获得一致入口通用默认实现减少接入工作高性能或供应商原生语义则保留给具体子类。另一方面batch、stream和ainvoke的存在不等于每个实现都以原生并行、细粒度 token 流或非阻塞 I/O 的方式运行。源码位置以下路径相对 LangChain 仓库根目录libs/core/langchain_core/runnables/base.pyRunnable定义统一执行契约及ainvoke、batch等默认行为RunnableLambda将 Python callable 包装为 RunnableRunnableSequence负责顺序执行各子步骤。6. 边界与错误认识.batch不等价于供应商批量 API。默认行为可能是并发多次 invoke。.stream有统一入口不保证组件每个都能产生细粒度 chunks。配置可沿组合流程传递但不应把敏感凭证写进 tags 或 metadata。Runnable 的类型提示和 JSON Schema 帮助表达契约但动态 Python 类型检查并不会因此自动覆盖所有运行时错误。小结Runnable 是 LangChain 框架“可以组合”的基础契约。它把单次执行、批处理、异步、流式和内省放到相似接口下子类可以继承默认行为也能按后端能力优化。本系列导航上一篇03 BaseChatModel 与 Provider 适配器——模型如何接入 LangChain下一篇05 LCEL 与 RunnableSequence、RunnableParallel 组合机制全系列LangChain 源码入门