深入学习和理解LangChain Runnables

1.LangChain Runnable介绍

LangChain的Runnable对象是一种协议(protocol)，它简化创建自定义链(chain)的过程。Runable是使用LangChain开发LLM应用程序的开发人员必须学习的一种抽象。

使用Ruanable协议可以将一系列的任务串联起来，将一个调用的输出作为输入提供给下一个，形成一个完整的流程。

Runnable协议通过提供一种标准化的方式使不同组件相互作用，简化了构建复杂工作流程的过程，使得通过一个函数调用，就可以执行一个链(chain)上的操作。

一个Runnable对象由以下部分组成：

方法：是Runnable对象可以执行的函数。Runnable对象的标准接口包括方法如invoke、batch、stream及其对应的异步方法ainvoke、abatch、astream。这些方法允许我们定义自定义链，并且可以以标准方式调用它们。
输入和输出schema：所有Runnable对象都给出了输入schema和输出schema，允许我们检查并理解Runnable对象期望的输入类型和它产生的输出类型。Runnable对象对象通过input_schema属性、output_schema属性和config_schema方法暴露有关其输入、输出和配置的模式信息。

LangChain可 Runnable对象提供了三种类型的方法来执行和与链交互：

invoke, ainvoke：使用单个输入执行可运行对象，通常用于处理单个数据片段时。
batch, abatch：允许并行处理多个输入。当有一系列输入并希望同时通过链运行它们时，此方法很有用。
stream, astream：将输入数据作为流进行处理，一次处理一个数据片段并提供可用结果。此方法非常适合处理实时数据处理或要增量处理的大型数据集的流式输出。

许多LangChain组件实现了Runnable协议，包括聊天模型chat models、LLMs、输出解析器output parsers、检索器retrievers、提示模板prompt templates等等。

可以使用管道|运算符将实现Runnable协议的组件连接起来，就可以创建一个复杂的自定义链，这让我们可以将LangChain的各个组件链接在一起以构建复杂的流程。

Runnable对象还对并发提供了支持，它的接口包括async方法，可以与asyncio的await语法一起用于并发。这允许我们同时运行多个任务，提高应用程序的效率和性能。

2.Runnable, RunnableLambda和Python Callable

在理解LangChain的Runnable前，先看一下Python中的Callable。

2.1 Python中的Callable

在Python中，Callable 是一个类型提示，用于表示可以被调用的对象。简单来说，任何可以调用的对象都是 Callable，包括函数、方法、类的实例（如果类定义了 __call__ 方法），甚至内置函数。

函数：Python 中的函数和方法都是 Callable，因为它们可以通过 () 来调用。

 def greet():
     print("Hello!")


 # greet 是一个 Callable
 greet()
 print(callable(greet))  # True

类和对象：如果一个类实现了 __call__ 方法，那么它的实例也会成为 Callable 对象。

 class Greeter:
     def __call__(self):
         print("Hello from instance!")

     # 这个类的实例也是一个 Callable

 greeter = Greeter()
 greeter()
 assert callable(greeter)

类型提示中的 Callable：在类型提示中，Callable 可以用来标注任何可以被调用的对象。通常，Callable 会用作类型注解，尤其是在函数的参数或返回值类型中。
```
 from collections.abc import Callable

 def execute_function(func: Callable[[], None]) -> None:
     func()

 def say_hello():
     print("Hello!")

 # say_hello 是一个 Callable
 execute_function(say_hello)
 assert callable(say_hello)
```
在上面的例子中，Callable[[], None] 表示一个不接受任何参数并且返回 None 的函数。

collections.abc.Callable[[参数类型1, 参数类型2], 返回类型] 可以指定函数的参数和返回类型。

from collections.abc import Callable

def add(x: int, y: int) -> int:
    return x + y

def operate(func: Callable[[int, int], int], a: int, b: int) -> int:
    return func(a, b)

# add 是一个 Callable
result = operate(add, 5, 3)  
print(result)
assert callable(add) # 8

lambda表达式创建匿名函数也是Callable对象

from collections.abc import Callable

def execute_function(func: Callable[[], None]) -> None:
    func()

# 使用 lambda 表达式，lambda 也是一个 Callable
execute_function(lambda: print("Hello from lambda!"))

2.2 使Python的Callable成为LangChain的Runnable

在 LangChain中，Runnable可以是任何Python中的Callable对象，但是，不能直接传递这些Callable对象，需要将它们包装在一个RunnableLambda对象中。

RunnableLambda将Python Callable对象转换为LangChain Runnable。

将Callable对象包装在RunnableLambda中使得该Callable对象可以在同步或异步上下文中使用。

RunnableLambda可以像其他Runnable对象一样进行组合，并提供与LangChain tracing的无缝集成。

RunnableLambda最适合不需要支持流式处理的代码。如果需要支持流式处理（即能够对输入块进行操作并产生输出块），需要使用RunnableGenerator。

from langchain_core.runnables import RunnableLambda


def add_one(x: int) -> int:
    return x + 1


async def add_one_async(x: int) -> int:
    return x + 1


async def main():
    runnable = RunnableLambda(add_one)

    print(runnable.invoke(1))  # 2
    print(runnable.batch([1, 2, 3]))  # [2, 3, 4]

    # Async is supported by default by delegating to the sync implementation
    print(await runnable.ainvoke(1))  # 2
    print(await runnable.abatch([1, 2, 3]))  # [2, 3, 4]

    # Alternatively, can provide both synd and async implementations
    runnable = RunnableLambda(add_one, afunc=add_one_async)
    print(runnable.invoke(1))  # Uses add_one
    print(await runnable.ainvoke(1))  # Uses add_one_async


import asyncio

if __name__ == "__main__":
    asyncio.run(main())

从上面的示例可以看出可以从Callable对象、异步Callable对象或两者一起创建RunnableLambda。

2.3 使用`@chain`装饰器创建Runnable对象

还可以通过@chain装饰器将任意函数转换为链。这在功能上等同于用RunnableLambda包装。

下面的示例，使用@chain装饰器创建一个自定义链，它结合了多个组件，例如提示、模型和输出解析器：

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import chain
from langchain_openai import ChatOpenAI

prompt1 = ChatPromptTemplate.from_template("Tell me a joke about {topic}")
prompt2 = ChatPromptTemplate.from_template("What is the subject of this joke: {joke}")

@chain
def custom_chain(text):
    prompt_val1 = prompt1.invoke({"topic": text})
    output1 = ChatOpenAI().invoke(prompt_val1)
    parsed_output1 = StrOutputParser().invoke(output1)
    chain2 = prompt2 | ChatOpenAI() | StrOutputParser()
    return chain2.invoke({"joke": parsed_output1})

custom_chain 现在是一个可运行对象，这意味可以对其使用 invoke。

custom_chain.invoke("bears")

3.RunnableSequence和"|“运算符

RunnableSequence是LangChain中最重要的组合运算符，因为它几乎在每个链中都会使用。

RunnableSequence可以直接实例化，或者更常见的是通过使用|运算符，其中左操作数或右操作数（或两者）必须是一个Runnable对象。

所有RunnableSequence对象都支持同步sync、异步async和批处理batch。batch和abatch的默认实现是利用线程池和asyncio gather，对于IO密集型的可运行对象来说，会比单纯调用invoke或ainvoke更快。

batch是通过按顺序调用RunnableSequence的每个组件的batch方法来实现的。

RunnableSequence本身也是一个Runnable行对象，可以继续被管道化。

下面是一个RunnableSequence的基本使用示例：

from langchain_core.runnables import RunnableLambda
from langchain_core.runnables import RunnableSequence


def add_one(x: int) -> int:
    return x + 1


def mul_two(x: int) -> int:
    return x * 2


async def main():
    runnable_1 = RunnableLambda(add_one)
    runnable_2 = RunnableLambda(mul_two)
    sequence = runnable_1 | runnable_2
    # sequence = RunnableSequence(first=runnable_1, last=runnable_2)
    print(sequence.invoke(1))  # 4
    print(await sequence.ainvoke(1))  # 4

    print(sequence.batch([1, 2, 3]))
    print(await sequence.abatch([1, 2, 3]))


import asyncio

if __name__ == "__main__":
    asyncio.run(main())

这里有一个使用LLM生成的JSON输出流的示例：

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers.json import SimpleJsonOutputParser
from langchain_openai import ChatOpenAI


async def main():
    prompt = PromptTemplate.from_template(
        "In JSON format, give me a list of {topic} and their "
        "corresponding names in French, Spanish and in a "
        "Cat Language."
    )

    model = ChatOpenAI(model="gpt-4o-mini")
    chain = prompt | model | SimpleJsonOutputParser()

    async for chunk in chain.astream({"topic": "colors"}):
        print("-")  # noqa: T201
        print(chunk, sep="", flush=True)  # noqa: T201


import asyncio

if __name__ == "__main__":
    asyncio.run(main())

“|“运算符是如何实现的

LangChain 中的Runnable对象是通过实现__or__运算符来支持|语法的。这在Runnable的源码中可以找到：

class Runnable(Generic[Input, Output], ABC):
...
    def __or__(
        self,
        other: Union[
            Runnable[Any, Other],
            Callable[[Any], Other],
            Callable[[Iterator[Any]], Iterator[Other]],
            Mapping[str, Union[Runnable[Any, Other], Callable[[Any], Other], Any]],
        ],
    ) -> RunnableSerializable[Input, Other]:
        """Compose this Runnable with another object to create a RunnableSequence."""
        return RunnableSequence(self, coerce_to_runnable(other))
...

4.RunnableParallel

RunnableParallel并行运行Runnables对象的字典，并返回其输出的字典。

RunnableParallel是LCEL的两个主要组合原语之一，另一个是前面介绍的RunnableSequence。

RunnableParallel并行调用Runnables对象，并为每个Runnables对象提供相同的输入。RunnableParallel可以直接实例化，也可以通过在RunnableSequence中使用字典字面量来实例化。RunnableSequence中的Mapping[str, Runnable[Input, Any]]会被转换为一个RunnableParallel。

from langchain_core.runnables import RunnableLambda


def add_one(x: int) -> int:
    return x + 1


def mul_two(x: int) -> int:
    return x * 2


def mul_three(x: int) -> int:
    return x * 3


async def main():
    runnable_1 = RunnableLambda(add_one)
    runnable_2 = RunnableLambda(mul_two)
    runnable_3 = RunnableLambda(mul_three)

    sequence = runnable_1 | {  # this dict is coerced to a RunnableParallel
        "mul_two": runnable_2,
        "mul_three": runnable_3,
    }
    # Or equivalently:
    # sequence = runnable_1 | RunnableParallel(
    #     {"mul_two": runnable_2, "mul_three": runnable_3}
    # )
    # Also equivalently:
    # sequence = runnable_1 | RunnableParallel(
    #     mul_two=runnable_2,
    #     mul_three=runnable_3,
    # )

    print(sequence.invoke(1))  # {'mul_two': 4, 'mul_three': 6}
    print(await sequence.ainvoke(1))  # {'mul_two': 4, 'mul_three': 6}

    print(
        sequence.batch([1, 2, 3])
    )  # [{'mul_two': 4, 'mul_three': 6}, {'mul_two': 6, 'mul_three': 9}, {'mul_two': 8, 'mul_three': 12}]
    print(
        await sequence.abatch([1, 2, 3])
    )  # [{'mul_two': 4, 'mul_three': 6}, {'mul_two': 6, 'mul_three': 9}, {'mul_two': 8, 'mul_three': 12}]


import asyncio

if __name__ == "__main__":
    asyncio.run(main())

RunnableParallel使并行Runnable变得简单。在下面的示例中，我们同时从两个不同的Runnable流式传输输出：

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableParallel
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")
joke_chain = ChatPromptTemplate.from_template("tell me a joke about {topic}") | model
poem_chain = (
    ChatPromptTemplate.from_template("write a 2-line poem about {topic}") | model
)

runnable = RunnableParallel(joke=joke_chain, poem=poem_chain)

# Display stream
output = {key: "" for key, _ in runnable.output_schema()}
print(output)
for chunk in runnable.stream({"topic": "bear"}):
    for key in chunk:
        output[key] = output[key] + chunk[key].content
    print(output)  # noqa: T201

5.RunnablePassthrough

RunnablePassthrough也是一个Runnable对象，它不会改变输入，但可以在输入是字典时，向输出添加额外的键。它经常与RunnableParallel一起使用，这允许我们在保持原始输入不变的同时添加一些额外信息。

示例1：

from langchain_core.runnables import RunnablePassthrough, RunnableParallel


runnable = RunnableParallel(
    {"context": lambda c: c.upper(), "question": RunnablePassthrough()}
)

print(runnable.invoke("what?"))  # {'context': 'WHAT?', 'question': 'what?'}

示例2:

from langchain_core.runnables import (
    RunnableLambda,
    RunnableParallel,
    RunnablePassthrough,
)

runnable = RunnableParallel(origin=RunnablePassthrough(), modified=lambda x: x + 1)

print(runnable.invoke(1))  # {'origin': 1, 'modified': 2}


def fake_llm(prompt: str) -> str:  # Fake LLM for the example
    return "completion"


chain = RunnableLambda(fake_llm) | {
    "original": RunnablePassthrough(),  # Original LLM output
    "parsed": lambda text: text[::-1],  # Parsing logic
}

print(chain.invoke("hello"))  # {'original': 'completion', 'parsed': 'noitelpmoc'}

在某些情况下，可能需要在将输入传递的同时向输出添加一些键。在这种情况下，您可以使用 assign 方法：

from langchain_core.runnables import RunnablePassthrough


def fake_llm(prompt: str) -> str:  # Fake LLM for the example
    return "completion"


runnable = {
    "llm1": fake_llm,
    "llm2": fake_llm,
} | RunnablePassthrough.assign(
    total_chars=lambda inputs: len(inputs["llm1"] + inputs["llm2"])
)

print(
    runnable.invoke("hello")
)  # {'llm1': 'completion', 'llm2': 'completion', 'total_chars': 20}

理解下面的代码：

...
# Retrieve and generate using the relevant snippets of the blog.
retriever = vectorstore.as_retriever()
prompt = hub.pull("rlm/rag-prompt")
# print(prompt)

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)
from langchain_core.runnables import  RunnableParallel
rag_chain = (
    RunnableParallel({"context": retriever | format_docs, "question": RunnablePassthrough()})
    | prompt
    | llm
    | StrOutputParser()
)

rag_chain.invoke("What is Task Decomposition?")

上面代码中prompt = hub.pull("rlm/rag-prompt")期望的输入是一个包含"context"和"question"键的字典，rag_chain中的RunnableParallel会从链的输入"What is Task Decomposition?"计算出这个字典。

6.RunnableGenerator

RunnableGenerator用于运行生成器函数。RunnableGenerator可以直接实例化，或者通过在RunnableSequence中使用生成器来实例化。

RunnableGenerator可以用来实现自定义行为，比如自定义输出解析器，同时保持流式处理能力。给定一个签名为Iterator[A] -> Iterator[B]的生成器函数，将其包装在RunnableGenerator中可以让它在从上一步流式传入输出块时立即发出这些块。

注意，如果生成器函数的签名为A -> Iterator[B]，即它需要上一步的输入完成才能发出块（例如，大多数LLM需要整个提示可用才能开始生成），那么它可以被包装在RunnableLambda中。

这里有一个例子来展示RunnableGenerator的基本机制：

from typing import Any, AsyncIterator, Iterator

from langchain_core.runnables import RunnableGenerator


def gen(input: Iterator[Any]) -> Iterator[str]:
    for token in ["Have", " a", " nice", " day"]:
        yield token


# Async version:
async def agen(input: AsyncIterator[Any]) -> AsyncIterator[str]:
    for token in ["Have", " a", " nice", " day"]:
        yield token


async def main():
    runnable = RunnableGenerator(gen)
    runnable.invoke(None)  # "Have a nice day"
    list(runnable.stream(None))  # ["Have", " a", " nice", " day"]
    runnable.batch([None, None])  # ["Have a nice day", "Have a nice day"]

    runnable = RunnableGenerator(agen)
    await runnable.ainvoke(None)  # "Have a nice day"
    [p async for p in runnable.astream(None)]  # ["Have", " a", " nice", " day"]


import asyncio

if __name__ == "__main__":
    asyncio.run(main())

RunnableGenerator使得在流式上下文中实现自定义行为变得容易:

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableGenerator, RunnableLambda
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from typing import Iterator


model = ChatOpenAI(model="gpt-4o-mini")
chant_chain = (
    ChatPromptTemplate.from_template("Give me a 3 word chant about {topic}")
    | model
    | StrOutputParser()
)


def character_generator(input: Iterator[str]) -> Iterator[str]:
    for token in input:
        if "," in token or "." in token:
            yield "👏" + token
        else:
            yield token


runnable = chant_chain | character_generator
assert type(runnable.last) is RunnableGenerator
print("".join(runnable.stream({"topic": "waste"})))  # Reduce👏, Reuse👏, Recycle👏.


# Note that RunnableLambda can be used to delay streaming of one step in a
# sequence until the previous step is finished:
def reverse_generator(input: str) -> Iterator[str]:
    # Yield characters of input in reverse order.
    for character in input[::-1]:
        yield character


runnable = chant_chain | RunnableLambda(reverse_generator)
print("".join(runnable.stream({"topic": "waste"})))  # ".elcycer ,esuer ,ecudeR"

1.LangChain Runnable介绍

2.Runnable, RunnableLambda和Python Callable

2.1 Python中的Callable

2.2 使Python的Callable成为LangChain的Runnable

2.3 使用@chain装饰器创建Runnable对象

3.RunnableSequence和"|“运算符

“|“运算符是如何实现的

4.RunnableParallel

5.RunnablePassthrough

6.RunnableGenerator

参考

2.3 使用`@chain`装饰器创建Runnable对象