LangChain 源码入门 04：Runnable 如何统一调用、批处理、异步与流式接口-开发者社区

1. 一个组件要具备哪些行为？

最简模型是：输入Input→ 处理 → 输出Output。如果每种组件都各自定义同步、异步、批量、流式的方法名，组合起来会很困难。Runnable 对这些执行方式提供统一词汇：

操作	用途
`invoke(x)`	处理一份输入
`ainvoke(x)`	在 async 程序中处理一份输入
`batch(xs)`	对多份输入进行批量/并发处理
`stream(x)`	随结果产生逐步返回输出

并非所有 Runnable 都原生支持高性能 async、批处理和流式。Base 类有通用实现，具体实现可按后端特性覆盖优化。

2. 从 Python 函数变成 Runnable

fromlangchain_core.runnablesimportRunnableLambdadefword_count(text:str)->int:returnlen(text.split())counter=RunnableLambda(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 runnable=RunnableLambda(lambdavalue:value*2)configured=runnable.with_config(tags=["lesson-04"],metadata={"feature":"demo"})print(configured.invoke(21))# 42schema=runnable.get_input_jsonschema()print(schema)

config里的 tags 和 metadata 可供执行上下文与观测系统使用。Runnable 还支持.with_retry(...)、.with_fallbacks(...)、.with_listeners(...)等统一包装，而不必让每个原始函数重新实现。

4. 异步调用示例

importasynciofromlangchain_core.runnablesimportRunnableLambda runnable=RunnableLambda(lambdavalue:value+1)asyncdefmain()->None:result=awaitrunnable.ainvoke(41)print(result)# 42asyncio.run(main())

这个示例可在 Python 脚本中运行，无需网络。默认ainvoke可将同步invoke放进 executor；provider-native async 实现可以覆盖它以避免阻塞线程池。

5. 源码设计解析：统一接口背后的默认实现

Runnable是抽象基类，不是一种会自动完成业务逻辑的对象。它定义单输入到单输出的核心操作invoke；组件继承这一接口，就能进入 LangChain 的组合与执行体系。比如RunnableLambda(word_count)包装器负责接收输入、调用 Python 函数并把结果作为输出返回。

@abstractmethoddefinvoke(self,input,config=None,**kwargs):"""Transform a single input into an output."""asyncdefainvoke(self,input,config=None,**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.py：Runnable定义统一执行契约及ainvoke、batch等默认行为；RunnableLambda将 Python callable 包装为 Runnable；RunnableSequence负责顺序执行各子步骤。