LangChain Runnables

在实际的应用场景中，输入、决策、输出过程是复杂、递进、动态的。可能需要路由、并行、串行、甚至这些机制的进一步组合。

Runnables 路由

一个常见的需求，是根据用户的输入动态决定一条由多个 Runnable 执行的 任务链。
决策的实现机制有两种：

RunnableLambda：运行任何的函数
RunnableBranch：基于用户输入动态路由

使用 RunnableLambda 进行动态路由

我们可以在管道上，使用任意被 RunnableLambda 函数。
需要注意的是，这里的函数必须是单个入参。如果你有多个入参，那需要进行柯里化。

def route(info):
    if "anthropic" in info["topic"].lower():
        return anthropic_chain
    elif "langchain" in info["topic"].lower():
        return langchain_chain
    else:
        return general_chain

full_chain = {"topic": chain, "question": lambda x: x["question"]} | RunnableLambda(
    route
)

RunnableLambda 可以接受一个 RunnableConfig。

使用 RunnableBranch 进行动态路由

由用户的输入，动态决定执行哪一条链。相比而言，通过 RunnableLambda 实现的路由策略可以更加灵活。
如下代码块中，接收一个用户输入。当输入中包含”anthropic”，则执行 anthropicChain；如果输入中包含 “langchain”，则执行 langChainChain；默认执行 generalChain。
感受下下面声明式的优雅逻辑写法。

const branch = RunnableBranch.from([
  [
    (x: { topic: string; question: string }) =>
      x.topic.toLowerCase().includes("anthropic"),
    anthropicChain,
  ],
  [
    (x: { topic: string; question: string }) =>
      x.topic.toLowerCase().includes("langchain"),
    langChainChain,
  ],
  generalChain,
]);

const fullChain = RunnableSequence.from([
  /*下面这个 Map，将会被自动转换为 RunnableMap。RunnableMap 所对应的两个键值，都是 Runnable 对象，两个 Runnable 对象并行执行。question 所对应的 Lambda 表达式被自动转换为 RunnableLambda */
  {
    topic: classificationChain,
    question: (input: { question: string }) => input.question,
  },
  branch,
]);

Runnables 并行

使用 RunnableMap

RunnableMap 允许可以并行执行多个 Runnable，并以 Map 的形式返回结果。

model = ChatOpenAI()
mapChain = RunnableMap.from({
  joke: PromptTemplate.fromTemplate("Tell me a joke about {topic}") | model,
  ),
  poem: PromptTemplate.fromTemplate("write a 2-line poem about {topic}") |
    new ChatAnthropic({}),,
});
await mapChain.invoke({ topic: "bear" });

RunnableMap 也等同于 RunnableParallel，是一种机制的两种名称。即 RunnableMap = RunnableParallel。

1
2
3

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

map_chain.invoke({"topic": "bear"})

使用 Runnable 的 batch

对于同一个 Runnable，也可以同时执行多个请求。
如下所示，max_concurrency 代表并发请求的最大个数。

1	chain.batch([{"topic": "bears"}, {"topic": "cats"}], config={"max_concurrency": 5})

Runnables 串行

RunnableSequence 可以允许多个 Runnable 被串行执行。串联起来的 Runnable，前一个 Runnable 的输出应是后一个 Runnable 的输入。当然，根据需求场景，串行任务和并行任务也可以进一步组合。

fullChain = RunnableSequence.from([
  {
    topic: classificationChain,
    question: (input: { question: string }) => input.question,
  },
  branch,
]);

参考链接

https://python.langchain.com/docs/expression_language/how_to/functions
https://agi-talks.vercel.app/202-langchain-chains/3
https://api.python.langchain.com/en/latest/core_api_reference.html