Sequential Chain

What is Sequential Chain?

A Sequential Chain is when you connect multiple Chains one after another — the output of Chain 1 becomes the input of Chain 2, and so on. Like an assembly line!

Input → [Chain 1] → output1 → [Chain 2] → output2 → [Chain 3] → Final Output

A single LLM Chain handles one task. But real problems often need multi-step reasoning:

Step	Task
1	Translate a user question to English
2	Answer the English question
3	Translate the answer back to Chinese

Execute multiple steps in sequence, where output of one step becomes input of the next step. Each chain has exactly one input and one output.

Output of chain N is automatically passed as input to chain N+1 Simple, linear, no variable naming needed. Each step depends on the previous — that’s a Sequential Chain.

Standard Environement Setup

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnablePassthrough

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
parser = StrOutputParser()

Part 1 — Linear Chain

Simplest case. One chain feeds into the next. Use lambda to repackage the string output into a dict for the next prompt.

prompt1 = ChatPromptTemplate.from_template(
    "Give me a creative company name for a company that sells {product}. "
    "Only return the name, nothing else."
)
prompt2 = ChatPromptTemplate.from_template(
    "Write a catchy one-line slogan for this company: {company_name}"
)
prompt3 = ChatPromptTemplate.from_template(
    "Write a short marketing email using this slogan: {slogan}"
)

chain1 = prompt1 | llm | parser
chain2 = prompt2 | llm | parser
chain3 = prompt3 | llm | parser

# ✅ Step by step, print input and output at every stage


# --- Step 1 ---
input_1 = {"product": "eco-friendly water bottles"}
output_1 = chain1.invoke(input_1)

print("=" * 50)
print("STEP 1")
print(f"  INPUT  : {input_1}")
print(f"  OUTPUT : {output_1}")
==================================================
STEP 1
  INPUT  : {"product": "eco-friendly water bottles"}
  OUTPUT : "AquaGreen Co."


# --- Step 2 ---
input_2 = {"company_name": output_1}
output_2 = chain2.invoke(input_2)

print("=" * 50)
print("STEP 2")
print(f"  INPUT  : {input_2}")
print(f"  OUTPUT : {output_2}")
==================================================
STEP 2
  INPUT  : {"company_name": "AquaGreen Co."}
  OUTPUT : "Drink Pure. Live Green."


# --- Step 3 ---
input_3 = {"slogan": output_2}
output_3 = chain3.invoke(input_3)

print("=" * 50)
print("STEP 3")
print(f"  INPUT  : {input_3}")
print(f"  OUTPUT : {output_3}")
==================================================
STEP 3
  INPUT  : {"slogan": "Drink Pure. Live Green."}
  OUTPUT : "Dear Customer, at AquaGreen Co. we believe..."



print("=" * 50)
print("FINAL OUTPUT:")
print(output_3)
==================================================
FINAL OUTPUT:
"Dear Customer, at AquaGreen Co. we believe..."

Part 2 — Parallel Chain

Run multiple chains on the same input simultaneously, then combine all results and pass them together to the next step.
对同一个输入同时运行多个 chain，把所有结果合并，一起传给下一步。

prompt_summary = ChatPromptTemplate.from_template(
    "Summarize this customer review in one sentence:\n{review}"
)
prompt_sentiment = ChatPromptTemplate.from_template(
    "Detect the sentiment of this review (Positive / Negative / Neutral):\n{review}"
)
prompt_reply = ChatPromptTemplate.from_template(
    """
    You are a customer service agent.
    Review summary : {summary}
    Sentiment      : {sentiment}
    Write a polite and helpful reply addressing their feedback.
    """
)

# ✅ Step by step debug
# Step 1 : Run parallel chains separately to see each output 

review = "The product broke after 2 days. Very disappointed with the quality."

input_1 = {"review": review}

output_summary   = (prompt_summary   | llm | parser).invoke(input_1)
output_sentiment = (prompt_sentiment | llm | parser).invoke(input_1)

print("=" * 50)
print("STEP 1a — Summary Chain")
print(f"  INPUT  : {input_1}")
print(f"  OUTPUT : {output_summary}")

==================================================
STEP 1a — Summary Chain
  INPUT  : {"review": "The product broke after 2 days..."}
  OUTPUT : "Customer reports product failure within 2 days."


print("=" * 50)
print("STEP 1b — Sentiment Chain")
print(f"  INPUT  : {input_1}")
print(f"  OUTPUT : {output_sentiment}")

==================================================
STEP 1b — Sentiment Chain
  INPUT  : {"review": "The product broke after 2 days..."}
  OUTPUT : "Negative"



# --- Step 2 : Feed combined results into reply chain ---
# --- 步骤2 : 把合并结果传入回复 chain ---
input_2 = {
    "summary"  : output_summary,
    "sentiment": output_sentiment,
}
output_reply = (prompt_reply | llm | parser).invoke(input_2)

print("=" * 50)
print("STEP 2 — Reply Chain")
print(f"  INPUT  : {input_2}")
print(f"  OUTPUT : {output_reply}")

==================================================
STEP 2 — Reply Chain
  INPUT  : {
      "summary"  : "Customer reports product failure within 2 days.",
      "sentiment": "Negative"
  }
  OUTPUT : "Dear Customer, we sincerely apologize..."


print("=" * 50)
print("FINAL OUTPUT / 最终输出:")
print(output_reply)

==================================================
FINAL OUTPUT / 最终输出:
"Dear Customer, we sincerely apologize..."

Part 3 — RunnablePassthroug

When a later step needs the original input AND the processed results, use RunnablePassthrough to carry the original input forward unchanged.
当后面的步骤既需要原始输入、又需要处理结果时，用 RunnablePassthrough 把原始输入原封不动地带过去。

# ✅ Run the full parallel step including Passthrough, inspect all keys


parallel_step = RunnableParallel(
    summary   = prompt_summary   | llm | parser,        # <-- Task A
    sentiment = prompt_sentiment | llm | parser,        # <-- Task B
    review    = RunnablePassthrough(),                  # <-- Task C
)

input_1 = {"review": "The product broke after 2 days. Very disappointed."}
intermediate = parallel_step.invoke(input_1)

print("=" * 50)
print("PARALLEL STEP — All outputs / 所有输出:")
print(f"  INPUT : {input_1}") 
# INPUT : {"review": "The product broke after 2 days. Very disappointed."}

print(f"  OUTPUT[review]     : {intermediate['review']}")
print(f"  OUTPUT[summary]    : {intermediate['summary']}")
print(f"  OUTPUT[sentiment]  : {intermediate['sentiment']}")
print("=" * 50)



#entitle output:
==================================================
PARALLEL STEP — All outputs / 所有输出:
  INPUT             : {"review": "The product broke after 2 days..."}
  OUTPUT[review]    : {"review": "The product broke after 2 days..."}
  OUTPUT[summary]   : "Customer reports product failure within 2 days."
  OUTPUT[sentiment] : "Negative"
==================================================

Step into each of them

# 先看这段代码在做什么

parallel_step = RunnableParallel(
    summary   = prompt_summary   | llm | parser,    # 任务A
    sentiment = prompt_sentiment | llm | parser,    # 任务B
    review    = RunnablePassthrough(),              # 任务C
)



 # 任务A
summary = prompt_summary | llm | parse
# prompt_summary 收到 input_1，填入模板
prompt_summary.invoke(input_1)
# → "Summarize this customer review in one sentence:                       # <-- prompt_summar
#    The product broke after 2 days. Very disappointed."

# llm 收到上面的 prompt，回复
llm.invoke(...)
# → AIMessage(content="Customer reports product failure within 2 days.")

# parser 提取纯字符串
parser.invoke(...)
# → "Customer reports product failure within 2 days."

Step 1 :

# Step 1
input_1 = {"review": "The product broke after 2 days. Very disappointed."}
print(input_1)
# → {"review": "The product broke after 2 days. Very disappointed."}

Step 2: parallel_step received input_1, it parallelly does 3 tasks (at the same time)

# Step 2  parallel_step received input_1, it parallelly does 3 tasks (at the same time)
intermediate = parallel_step.invoke(input_1)

parallel_step = RunnableParallel(
    summary   = ...,   # 任务A
    sentiment = ...,   # 任务B
    review    = ...,   # 任务C
)

Step 3 : Task A

 # Task A
summary = prompt_summary | llm | parse

# prompt_summary 收到 input_1，fill in template
prompt_summary.invoke(input_1)
# → "Summarize this customer review in one sentence:
#    The product broke after 2 days. Very disappointed."

# llm get above prompt，respone.
llm.invoke(...)
# → AIMessage(content="Customer reports product failure within 2 days.")

# parser extract pure words 
parser.invoke(...)
# → "Customer reports product failure within 2 days."

Task A output： "Customer reports product failure within 2 days."

Step 4 : Task B

# Task B
sentiment = prompt_sentiment | llm | parse

# received input_1 too
prompt_sentiment.invoke(input_1)
# → "Detect the sentiment of this review (Positive/Negative/Neutral):
#    The product broke after 2 days. Very disappointed."

llm.invoke(...)
# → AIMessage(content="Negative")

parser.invoke(...)
# → "Negative"

Task A output： "Negative"

Step 5: Task C

# Step 5: Task C
review = RunnablePassthrough()

# RunnablePassthrough does nothing 
# 什么都不做, 原封不动把 input_1 传过去
# → {"review": "The product broke after 2 days. Very disappointed."}

Task C output: {"review": "The product broke after 2 days. Very disappointed."}

Step 6 — RunnableParallel

Once all three tasks finish running, RunnableParallel, Merge Tasks A, B, and C, packages the results into a dictionary.”
三个任务都跑完之后，RunnableParallel 把三个结果合并打包成一个 dict：

intermediate = {
    "summary"  : "Customer reports product failure within 2 days.",   # Task A
    "sentiment": "Negative",                                          # Task B
    "review"   : {"review": "The product broke after 2 days..."},     # Task C
}

Step 7: print out each rows

print(f"  INPUT              : {input_1}")
# → INPUT : {"review": "The product broke after 2 days..."}
#   就是你最开始传进去的原始 dict

print(f"  OUTPUT[review]     : {intermediate['review']}")
# → OUTPUT[review] : {"review": "The product broke after 2 days..."}
#   RunnablePassthrough 原封不动的输出，和 input_1 完全一样

print(f"  OUTPUT[summary]    : {intermediate['summary']}")
# → OUTPUT[summary] : "Customer reports product failure within 2 days."
#   LLM 生成的摘要

print(f"  OUTPUT[sentiment]  : {intermediate['sentiment']}")
# → OUTPUT[sentiment] : "Negative"
#   LLM 判断的情感