زنجیره‌های ترتیبی، موازی و مسیریاب

خیلی وقت‌ها یک کار با یک فراخوانی مدل تمام نمی‌شود: اول خلاصه کن، بعد ترجمه کن؛ یا هم‌زمان مزایا، معایب و خلاصه‌ی یک محصول را بنویس؛ یا اول ببین سؤال درباره‌ی چیست و بعد به متخصص مربوط بده. در دوره‌ی اول خروجی یک مرحله را دستی به مرحله‌ی بعد دادیم. در این درس همین کارها را با LCEL به شکل زنجیره درمی‌آوریم.

زنجیره‌ی ترتیبی#

دو زنجیره داریم: یکی متن را در یک جمله خلاصه می‌کند و دیگری ترجمه می‌کند.

from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

llm = init_chat_model("gpt-5.2", model_provider="openai", temperature=0)

long_text = """LangChain is a framework for developing applications powered by large language models.
It provides tools and abstractions to improve the customization, accuracy, and relevancy of the information
the models generate. It includes APIs and integrations for a wide variety of components, including vector
stores, document loaders, and output parsers."""

summarize_prompt = ChatPromptTemplate.from_messages([
    ("system", "Summarize the following text in one sentence. with no details"),
    ("human", "{text}"),
])
summarize_chain = summarize_prompt | llm | StrOutputParser()

translate_prompt = ChatPromptTemplate.from_messages([
    ("system", "Translate the following text to Persian. Return only the translated text and nothing else. "
               "Do not add explanations, notes, or extra words."),
    ("human", "{text}"),
])
translate_chain = translate_prompt | llm | StrOutputParser()

sequential_chain = summarize_chain | translate_chain
sequential_chain.invoke(long_text)
# 'لانگ‌چین یک چارچوب برای ساخت برنامه‌هایی است که توسط مدل‌های زبانی بزرگ پشتیبانی می‌شوند.'

این کار می‌کند، چون خروجی زنجیره‌ی اول یک رشته است و قالب دوم فقط یک متغیر دارد. همان‌طور که در درس قبل دیدیم، چنین قالبی رشته‌ی خام را هم می‌پذیرد.

وقتی ورودی و خروجی جور نیستند#

حالا زبان مقصد را هم متغیر کنیم: "Translate the following text to {lang}.". قالب دوم دو متغیر دارد (text و lang)، اما زنجیره‌ی اول فقط یک رشته تحویل می‌دهد. نتیجه خطاست، چون LangChain نمی‌داند این رشته جای کدام متغیر بنشیند.

راه‌حل یک رابط بین دو زنجیره است. RunnableLambda هر تابع پایتون را به Runnable تبدیل می‌کند تا بشود آن را وسط زنجیره گذاشت:

from langchain_core.runnables import RunnableLambda

sequential_chain = (
    summarize_chain
    | RunnableLambda(lambda x: {"text": x, "lang": "Persian"})   # رشته → دیکشنری‌ای که قالب دوم می‌خواهد
    | translate_chain
)
result = sequential_chain.invoke({"text": long_text})

LangChain یک چارچوب برای ساخت برنامه‌های مبتنی بر مدل‌های زبانی بزرگ است که ابزارها، انتزاع‌ها و
یکپارچه‌سازی‌ها (مانند ذخیره‌سازهای برداری، بارگذارهای اسناد و تجزیه‌گرهای خروجی) را ارائه می‌دهد …

lambda x: … یک تابع بی‌نام است که خروجی مرحله‌ی قبل (x) را می‌گیرد و به شکل لازم برای مرحله‌ی بعد درمی‌آورد. این خلاصه از نسخه‌ی قبلی بلندتر شد، چون در این اجرا عبارت «with no details» را از پرامپت خلاصه‌ساز برداشته بودیم. دقیقاً همان اثری که در درس توسعه‌ی تکرارشونده‌ی پرامپت دیدیم.

زنجیره‌ی موازی#

برای یک محصول سه چیز می‌خواهیم: سه مزیت، سه عیب و یک خلاصه‌ی یک‌خطی. این سه کار به هم وابسته نیستند، پس دلیلی ندارد منتظر هم بمانند:

from langchain_core.runnables import RunnableParallel

pros_prompt    = ChatPromptTemplate.from_messages([("system", "List 3 pros of the given product in bullet points."), ("human", "{product}")])
cons_prompt    = ChatPromptTemplate.from_messages([("system", "List 3 cons of the given product in bullet points."), ("human", "{product}")])
summary_prompt = ChatPromptTemplate.from_messages([("system", "Write a one-line summary of the given product."), ("human", "{product}")])

parallel_chain = RunnableParallel(
    pros=pros_prompt | llm | StrOutputParser(),
    cons=cons_prompt | llm | StrOutputParser(),
    summary=summary_prompt | llm | StrOutputParser(),
)

result = parallel_chain.invoke({"product": "iPhone 15 Pro"})
print(result["summary"])   # iPhone 15 Pro is Apple's premium smartphone featuring a titanium design, A17 Pro chip, …
print(result["pros"])
print(result["cons"])

RunnableParallel ورودی را به همه‌ی شاخه‌ها می‌دهد، آن‌ها را هم‌زمان اجرا می‌کند و یک دیکشنری با همان کلیدها برمی‌گرداند. این دیکشنری می‌تواند ورودی یک زنجیره‌ی دیگر باشد، مثلاً قالبی که {pros} و {cons} را می‌گیرد و یک بررسی کامل می‌نویسد.

چقدر سریع‌تر؟#

سه سؤال کوتاه با جواب یک‌کلمه‌ای (پایتخت ایران، تعداد پاهای سگ، رنگ سیب):

یعنی حدود ۲٫۸ برابر سریع‌تر. هر چه شاخه‌ها بیشتر و کندتر باشند، صرفه‌جویی بیشتر است. اینجا هم مثل batch، با Ollama روی یک کارت گرافیک خانگی این افزایش سرعت کمتر دیده می‌شود.

زنجیره‌ی مسیریاب#

مثل تلفن پشتیبانی یک شرکت: اول می‌پرسد سؤالتان درباره‌ی چیست و بعد شما را به واحد مربوط وصل می‌کند. اینجا یک فراخوانی مدل هماهنگ‌کننده است و موضوع سؤال را تشخیص می‌دهد. خروجی‌اش باید دقیقاً یکی از چند مقدار ثابت باشد، پس از خروجی ساخت‌یافته با Literal استفاده می‌کنیم:

from typing import Literal
from pydantic import BaseModel

class RouteQuery(BaseModel):
    """تشخیص موضوع سوال"""
    topic: Literal["math", "history", "science", "other"]

router_llm = llm.with_structured_output(RouteQuery)

def route(input_dict):
    topic = router_llm.invoke(input_dict["question"]).topic
    print(f"→ Routed to: {topic}")
    chains = {
        "math":    math_prompt    | llm | StrOutputParser(),   # "You are a math expert. Answer step by step."
        "history": history_prompt | llm | StrOutputParser(),   # "You are a history expert. Provide historical context."
        "science": science_prompt | llm | StrOutputParser(),
        "other":   general_prompt | llm | StrOutputParser(),
    }
    return chains[topic].invoke(input_dict)

Q: What is the derivative of x^2?          → Routed to: math
Q: When did World War II end?              → Routed to: history
Q: What is the best programming language?  → Routed to: other
Q: How does photosynthesis work?           → Routed to: science

Literal تضمین می‌کند که topic فقط یکی از همین چهار مقدار باشد. پس chains[topic] هیچ‌وقت KeyError نمی‌دهد. اگر مسیریاب را با RunnableLambda(route) بپیچید، خودش هم یک Runnable می‌شود و می‌تواند جزئی از یک زنجیره‌ی بزرگ‌تر باشد.

کاربردهای واقعی: فرستادن سؤال ساده به مدل ارزان و سؤال سخت به مدل قوی، انتخاب پرامپت متخصص برای هر حوزه، یا جدا کردن پیام‌های شکایت، سؤال فنی و درخواست بازگشت وجه.

خلاصه#

• زنجیره‌ها را می‌شود با | پشت سر هم گذاشت. اگر خروجی یکی با ورودی بعدی جور نیست، RunnableLambda رابط می‌شود.
• RunnableParallel شاخه‌های مستقل را هم‌زمان اجرا می‌کند و دیکشنری برمی‌گرداند. در مثال ما ۲٫۸ برابر سریع‌تر بود.
• مسیریاب با with_structured_output و Literal موضوع را تشخیص می‌دهد و زنجیره‌ی مناسب را اجرا می‌کند.