怎么精确控制RunnableParallel.assign中Runnable的并行或串行执行。

[shell-nlp]

Checked other resources

• I added a very descriptive title to this question.
• I searched the LangChain documentation with the integrated search.
• I used the GitHub search to find a similar question and didn't find it.

Commit to Help

• I commit to help with one of those options 👆

Example Code

qa_bot_with_docs = RunnableParallel(
    {
        "context": itemgetter("question") | contextual_rerank_retriever,
        "question": itemgetter("question"),
    }
).assign(
    answer=qa_bot,
    prompt=prompt,
    # 能否将event、org、human设置为串行处理，因为 并行会导致 单个chain 耗时增加，影响用户体验
    event=event_extraction_chain,
    org=org_extraction_chain,
    human=human_extraction_chain,
)

Description

能否将event、org、human设置为串行处理，因为 并行会导致 单个chain 耗时增加，影响用户体验
怎么控制RunnableParallel.assign 中 runnable 是并行执行还是串行执行。最好可以精确控制某几个chain的并行或串行。

System Info

System Information

OS: Linux
OS Version: #101~20.04.1-Ubuntu SMP Thu Nov 16 14:22:28 UTC 2023
Python Version: 3.10.9 (main, Mar 1 2023, 18:23:06) [GCC 11.2.0]

Package Information

langchain_core: 0.2.28
langchain: 0.2.9
langchain_community: 0.2.7
langsmith: 0.1.85
langchain_experimental: 0.0.62
langchain_openai: 0.1.7
langchain_text_splitters: 0.2.2
langchainhub: 0.1.14
langgraph: 0.1.8

Packages not installed (Not Necessarily a Problem)

The following packages were not found:

langserve

[dosubot[bot]]

要精确控制 RunnableParallel.assign 中 Runnable 的并行或串行执行，特别是某几个 chain 的执行方式，可以通过将需要串行执行的部分单独定义为一个 RunnableSequence，然后在 RunnableParallel 中组合这些 Runnable。以下是一个示例代码：

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

model = ChatOpenAI()

# 定义单独的 Runnables
event_chain = ChatPromptTemplate.from_template("extract event from {text}") | model
org_chain = ChatPromptTemplate.from_template("extract organization from {text}") | model
human_chain = ChatPromptTemplate.from_template("extract human from {text}") | model

# 将需要串行执行的部分组合成一个 RunnableSequence
serial_chain = RunnableSequence(event_chain, org_chain, human_chain)

# 将串行和并行的部分组合在一起
qa_bot_with_docs = RunnableParallel(
    {
        "context": itemgetter("question") | contextual_rerank_retriever,
        "question": itemgetter("question"),
        "serial_chain": serial_chain  # 串行执行的部分
    }
).assign(
    answer=qa_bot,
    prompt=prompt
)

# 执行
qa_bot_with_docs.invoke({"question": "What is the event?"})

在这个示例中，event_chain、org_chain 和 human_chain 被组合成一个 RunnableSequence，确保它们串行执行。然后，这个 RunnableSequence 被包含在 RunnableParallel 中，以便与其他并行执行的 Runnable 一起运行^[1][2].

To continue talking to Dosu, mention @dosu.

[shell-nlp]

我的目的是在最终的ouput中要包含answer、event、org、human字段，且执行的顺序是 先执行answer，执行完成后再执行event、org、human

[dosubot[bot]]

要确保最终输出包含 answer、event、org 和 human 字段，并且执行顺序是先执行 answer，然后再执行 event、org 和 human，可以按照以下方式设置链：

from langchain_core.runnables import RunnableParallel, RunnableSequence, RunnablePassthrough

# 假设 `answer`, `event`, `org`, `human` 是已定义的 Runnables
# 例如：
# answer = ...
# event = ...
# org = ...
# human = ...

# 将 event, org, human 组合成一个串行执行的 RunnableSequence
serial_chain = RunnableSequence(event, org, human)

# 创建一个包含所有步骤的 RunnableParallel
chain = (
    RunnableParallel(question=RunnablePassthrough(), docs=wiki)
    .assign(answer=answer)
    .assign(serial_chain=serial_chain)
    .pick(["answer", "serial_chain.event", "serial_chain.org", "serial_chain.human"])
)

# 执行
result = chain.invoke({"question": "What is the event?"})
print(result)

在这个示例中，answer 会首先执行，然后 event、org 和 human 会按顺序串行执行。最终输出将包含 answer、event、org 和 human 字段 ^[1][2].

To continue talking to Dosu, mention @dosu.