Skip to main content

如何并行调用可运行程序

¥How to invoke runnables in parallel

Prerequisites

本指南假设你熟悉以下概念：

¥This guide assumes familiarity with the following concepts:

RunnableParallel（也称为 RunnableMap）原语是一个对象，其值为可运行对象（或可强制转换为可运行对象，例如函数）。它并行运行所有值，每个值都通过 RunnableParallel 的初始输入进行调用。最终返回值是一个对象，每个值的结果都包含在其相应的键下。

¥The RunnableParallel (also known as a RunnableMap) primitive is an object whose values are runnables (or things that can be coerced to runnables, like functions). It runs all of its values in parallel, and each value is called with the initial input to the RunnableParallel. The final return value is an object with the results of each value under its appropriate key.

使用 RunnableParallels 格式化​

¥Formatting with RunnableParallels

RunnableParallels 对于并行化操作很有用，但也可用于操作一个 Runnable 的输出以匹配序列中下一个 Runnable 的输入格式。你可以使用它们来拆分或分叉链，以便多个组件可以并行处理输入。之后，其他组件可以连接或合并结果以合成最终响应。这种类型的链会创建一个如下所示的计算图：

¥RunnableParallels are useful for parallelizing operations, but can also be useful for manipulating the output of one Runnable to match the input format of the next Runnable in a sequence. You can use them to split or fork the chain so that multiple components can process the input in parallel. Later, other components can join or merge the results to synthesize a final response. This type of chain creates a computation graph that looks like the following:

     Input
      / \
     /   \
 Branch1 Branch2
     \   /
      \ /
      Combine

下面，RunnableParallel 中每个链的输入预期为一个对象，其键为 "topic"。我们可以通过使用与该结构匹配的对象调用我们的链来满足该要求。

¥Below, the input to each chain in the RunnableParallel is expected to be an object with a key for "topic". We can satisfy that requirement by invoking our chain with an object matching that structure.

tip

请参阅 此部分包含有关安装集成包的一般说明。

¥See this section for general instructions on installing integration packages.

npm install @langchain/anthropic @langchain/cohere @langchain/core
import { PromptTemplate } from "@langchain/core/prompts";
import { RunnableMap } from "@langchain/core/runnables";
import { ChatAnthropic } from "@langchain/anthropic";

const model = new ChatAnthropic({});
const jokeChain = PromptTemplate.fromTemplate(
  "Tell me a joke about {topic}"
).pipe(model);
const poemChain = PromptTemplate.fromTemplate(
  "write a 2-line poem about {topic}"
).pipe(model);

const mapChain = RunnableMap.from({
  joke: jokeChain,
  poem: poemChain,
});

const result = await mapChain.invoke({ topic: "bear" });
console.log(result);
/*
  {
    joke: AIMessage {
      content: " Here's a silly joke about a bear:\n" +
        '\n' +
        'What do you call a bear with no teeth?\n' +
        'A gummy bear!',
      additional_kwargs: {}
    },
    poem: AIMessage {
      content: ' Here is a 2-line poem about a bear:\n' +
        '\n' +
        'Furry and wild, the bear roams free  \n' +
        'Foraging the forest, strong as can be',
      additional_kwargs: {}
    }
  }
*/

API Reference:

操作输出/输入​

¥Manipulating outputs/inputs

Map 可用于操作一个 Runnable 的输出，使其与序列中下一个 Runnable 的输入格式匹配。

¥Maps can be useful for manipulating the output of one Runnable to match the input format of the next Runnable in a sequence.

请注意，RunnableSequence.from() 调用中的对象会自动强制转换为可运行的 map。对象的所有键必须具有可运行的值，或者可以强制转换为可运行的值（函数转换为 RunnableLambda 或对象转换为 RunnableMap）。通过 .pipe() 方法组合链时也会发生这种强制转换。

¥Note below that the object within the RunnableSequence.from() call is automatically coerced into a runnable map. All keys of the object must have values that are runnables or can be themselves coerced to runnables (functions to RunnableLambdas or objects to RunnableMaps). This coercion will also occur when composing chains via the .pipe() method.

import { CohereEmbeddings } from "@langchain/cohere";
import { PromptTemplate } from "@langchain/core/prompts";
import { StringOutputParser } from "@langchain/core/output_parsers";
import {
  RunnablePassthrough,
  RunnableSequence,
} from "@langchain/core/runnables";
import { Document } from "@langchain/core/documents";
import { ChatAnthropic } from "@langchain/anthropic";
import { MemoryVectorStore } from "langchain/vectorstores/memory";

const model = new ChatAnthropic();
const vectorstore = await MemoryVectorStore.fromDocuments(
  [{ pageContent: "mitochondria is the powerhouse of the cell", metadata: {} }],
  new CohereEmbeddings({ model: "embed-english-v3.0" })
);
const retriever = vectorstore.asRetriever();
const template = `Answer the question based only on the following context:
{context}

Question: {question}`;

const prompt = PromptTemplate.fromTemplate(template);

const formatDocs = (docs: Document[]) => docs.map((doc) => doc.pageContent);

const retrievalChain = RunnableSequence.from([
  { context: retriever.pipe(formatDocs), question: new RunnablePassthrough() },
  prompt,
  model,
  new StringOutputParser(),
]);

const result = await retrievalChain.invoke(
  "what is the powerhouse of the cell?"
);
console.log(result);

/*
 Based on the given context, the powerhouse of the cell is mitochondria.
*/

API Reference:

这里，提示的输入应为包含键 "context" 和 "question" 的映射。用户输入就是问题。因此，我们需要使用检索器获取上下文，并将用户输入传递到 "question" 键下。

¥Here the input to prompt is expected to be a map with keys "context" and "question". The user input is just the question. So we need to get the context using our retriever and passthrough the user input under the "question" key.

后续步骤​

¥Next steps

现在你了解了一些使用 RunnableParallel 格式化和并行化链式步骤的方法。

¥You now know some ways to format and parallelize chain steps with RunnableParallel.

接下来，你可能会对链中的 使用自定义逻辑 感兴趣。

¥Next, you might be interested in using custom logic in your chains.