استفاده از LangChain Expression Language در پایتون — از پایه تا پیاده‌سازی

Gemini

2026/09/19

زمان مطالعه: 25 دقیقه

22 بازدید

استفاده از LangChain Expression Language در پایتون — از پایه تا پیاده‌سازی

مقدمه‌ای بر دردسرهای زنجیره‌سازی سنتی و ظهور LCEL

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

توسعه‌دهندگان مبتدی معمولاً توابع تودرتو یا کلاس‌های سفارشی سنگینی می‌نویسند که مدیریت خطا و استریمینگ در آن‌ها تقریباً غیرممکن است. در این روش سنتی، هر تابع به صورت دستی خروجی مدل قبلی را دریافت کرده و به ورودی بعدی تبدیل می‌کند که با کوچک‌ترین تغییر در ساختار داده، کل زنجیره متوقف می‌شود.

# رویکرد سنتی و شکننده
def run_pipeline(user_input):
    prompt = template.format(input=user_input)
    response = model.predict(prompt)
    return parser.parse(response)

زبان بیان LangChain یا همان LCEL یک رویکرد اعلامی بر پایه پروتکل استاندارد Runnable فراهم می‌کند تا قطعات کوچک به راحتی به هم متصل شوند. این معماری لوله‌کشی به ما اجازه می‌دهد اجزای مختلف سیستم را بدون نیاز به نوشتن کدهای چسبنده اضافه، به شکل زنجیره‌ای اجرا کنیم.

Input -> RunnablePrompt -> RunnableModel -> RunnableOutputParser -> Final Result

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

# ساختار بهینه با LCEL
chain = prompt | model | parser
result = chain.invoke({"topic": "پایتون"})

مقایسه روش‌های سنتی فراخوانی API با ساختار LCEL تفاوت چشمگیری در خوانایی و قابلیت مدیریت خطا ایجاد می‌کند. در روش سنتی، ما با انبوهی از شرایط شرطی و بلوک‌های Try-Except دست‌وپنجه نرم می‌کنیم که نگهداری پروژه را دشوار می‌سازد. در مقابل، LCEL با بهره‌گیری از عملگر | جریان داده را به صورت خطی و شفاف مدیریت می‌کند.

نکته حرفه‌ای: هیچ‌گاه منطق بیزینس پیچیده را مستقیماً داخل توابع پردازش پرامپت قرار ندهید؛ بلکه آن‌ها را به عنوان اجزای مستقل در زنجیره LCEL تعریف کنید تا قابلیت تست‌پذیری حفظ شود.

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

{
  "status": "success",
  "data": "نتیجه پردازش شده توسط مدل"
}

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

پایه‌ترین بلوک ساختاری: درک ساختار Runnable و متد invoke

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

برای حل این مشکل ساختاری، زبان LCEL بر پایه یک اینترفیس انتزاعی و قدرتمند به نام Runnable بنا شده است. انتخاب این الگو به این دلیل است که تمام اجزای سیستم، از ساده‌ترین تابع تبدیل داده تا پیچیده‌ترین مدل زبانی، از یک قرارداد رفتاری واحد پیروی می‌کنند و در نتیجه می‌توان آن‌ها را بدون نگرانی از جزئیات داخلی به یکدیگر متصل کرد.

Data Dictionary -> Runnable Interface -> Execution Context -> Processed Output

برای درک نحوه کارکرد این پروتکل، به جای نوشتن توابع پراکنده، از متد استاندارد invoke برای اجرای سنکرون و از ainvoke برای اجرای آسنکرون استفاده می‌کنیم. در ادامه ساختار یک شیء سفارشی که از این الگو پیروی می‌کند را در چند خط پیاده‌سازی می‌کنیم:

from langchain_core.runnables import RunnableLambda

def process_text(text: str) -> dict:
    return {"length": len(text), "content": text.upper()}

runnable_task = RunnableLambda(process_text)

در قطعه کد بالا، تابع RunnableLambda یک تابع ساده پایتون را به یک شیء استاندارد با قابلیت invoke تبدیل می‌کند. نوع ورودی این شیء یک رشته متنی است و خروجی آن در قالب یک دیکشنری ساختاریافته پایتون تنظیم شده است تا لایه‌های بعدی زنجیره دقیقاً بدانند با چه نوع داده‌ای روبه‌رو هستند. اگر بخواهیم این وظیفه را اجرا کنیم، متد invoke را با یک دیکشنری یا مقدار ورودی صدا می‌زنیم:

result = runnable_task.invoke("سلام لنگ‌چین")
print(result)

اجرای این کد ساختار استاندارد زیر را به عنوان خروجی JSON بازمی‌گرداند که کاملاً قابل پیش‌بینی و آماده استفاده در لایه بیزینس است:

{
  "length": 13,
  "content": "سلام لنگ‌چین"
}

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

نکته کاربردی: فراموش کردن ارسال ساختار داده صحیح یا عدم رعایت نوع کلیدها هنگام فراخوانی متد invoke خطای KeyError ایجاد می‌کند. همیشه مطمئن شوید ورودی ارسالی با انتظارات ساختاری آن بلوک همخوانی دارد.

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

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

validator = RunnableLambda(lambda x: {"valid": len(x) > 5})
print(validator.invoke("تست"))

خروجی اجرای این نوع دوم به صورت زیر خواهد بود:

{
  "valid": false
}

عدد 5 در شرط طول متن به این دلیل انتخاب شده است که درخواست‌های بسیار کوتاه و ناقص پیش از مصرف منابع گران‌قیمت شبکه فیلتر شوند. تغییر این مقدار مستقیماً روی سخت‌گیری لایه اعتبارسنجی تأثیر می‌گذارد.

در این بخش آموختیم چگونه با استفاده از اینترفیس Runnable و متد invoke یکپارچگی جریان داده را از همان پایه‌ترین بلوک ساختاری تضمین کنیم.

اتصال زنجیره‌ها با عملگر قدرتمند لوله (|)

هنگامی که پروژه‌های هوش مصنوعی مقیاس‌پذیر می‌شوند، نوشتن کدهای طولانی برای پاس دادن صریح خروجی یک ماژول به ورودی ماژول بعدی، خوانایی کد را به شدت کاهش می‌دهد و نگهداری سیستم را دشوار می‌کند.

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

راهکار مهندسی انتخاب‌شده در فریم‌ورک LangChain، استفاده از عملگر | (Bitwise OR) برای بازنویسی منطق به صورت خواناترین شکل ممکن است که دقیقاً مشابه لوله‌کشی در محیط لینوکس عمل می‌کند و جریان داده را بین اشیاء Runnable هدایت می‌کند.

سربار عملگر | در پایتون با بازنویسی متدهای جادویی (or) در کلاس پایه Runnable مدیریت می‌شود تا اشیاء بتوانند بدون نیاز به کدهای واسط، به یکدیگر متصل شوند و یک ساختار یکپارچه تشکیل دهند.

PromptTemplate -> | -> ChatModel -> | -> StrOutputParser

ترتیب قرارگیری این کامپوننت‌ها کاملاً حیاتی است و داده‌ها به صورت خطی از چپ به راست جریان پیدا می‌کنند تا پردازش نهایی انجام شود.

در ادامه، پیاده‌سازی این زنجیره سه مرحله‌ای را در کمتر از ۱۰ خط مشاهده می‌کنید:

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_template("یک جوک کوتاه درباره {topic} بگو.")
model = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

chain = prompt | model | parser
response = chain.invoke({"topic": "برنامه‌نویسی"})
print(response)

در این ساختار، شیء prompt قالب متن را آماده می‌کند، شیء model درخواست را به سمت مدل زبانی هدایت می‌کند و شیء parser شیء خروجی را به یک رشته متنی خالص تبدیل می‌کند.

نتیجه نهایی متنی که مستقیماً از پارسر خارج می‌شود به شکل زیر خواهد بود:

برنامه‌نویس‌ها چای را سرد می‌نوشند چون کدهایشان به اندازه کافی داغ است!

نکته: عملگر | از چپ به راست داده‌ها را هدایت می‌کند و ترتیب قرارگیری کامپوننت‌ها حیاتی است؛ اگر پارسر را در ابتدا قرار دهید، به دلیل عدم تطابق نوع ورودی با خطاهای زمان اجرا مواجه خواهید شد.

برای بررسی تغییرات رفتاری، جابجایی ترتیب اجزا را در حالت‌های مختلف تست می‌کنیم تا تاثیر خطاهای زمان اجرا را مشاهده کنیم.

حالت اول (model | prompt | parser): با خطای عدم تطابق نوع داده مواجه می‌شویم زیرا مدل زبانی انتظار ورودی متنی یا پرامپت دارد نه شیء خام.
حالت دوم (prompt | parser | model): پارسر خروجی پرامپت را به رشته تبدیل می‌کند و مدل زبانی به جای ساختار پیام، یک رشته ساده دریافت می‌کند که دقت پاسخ را کاهش می‌دهد.

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

مدیریت پرامپت‌های پویا و تزریق متغیرها در LCEL

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

یک توسعه‌دهنده جونیور معمولاً برای حل این مسئله از متدهای فرمت‌دهی رشته‌ای استاندارد در پایتون مانند f-string استفاده می‌کند. این رویکرد به ظاهر کار را راه می‌اندازد، اما با ورود داده‌های مخرب یا کاراکترهای خاص از سمت کاربر، خطاهای امنیتی و تزریق پرامپت رخ می‌دهد. همچنین این روش رشته ساده‌ای تولید می‌کند که قابلیت مدیریت ساختار پیام‌ها در مدل‌های گفتگو‌محور را از بین می‌برد.

برای رفع این مشکل در معماری LCEL، از ChatPromptTemplate به عنوان یک شیء قابل اجرا استفاده می‌کنیم. این کلاس به ما اجازه می‌دهد پرامپت‌ها را به صورت قالب‌های ساختاریافته تعریف کنیم تا متغیرها به صورت ایمن به آن‌ها تزریق شوند.

User Input -> ChatPromptTemplate -> Formatted Messages -> LLM

این ساختار باعث می‌شود قالب پرامپت به عنوان یک گره مستقل در خط لوله عمل کند. مقادیر ورودی مستقیماً به عنوان پارامترهای زنجیره ارسال می‌شوند و نیازی به پردازش دستی رشته‌ها نیست.

در ادامه، نحوه تعریف یک قالب پرامپت پویا با نقش‌های سیستم و کاربر را مشاهده می‌کنید:

from langchain_core.prompts import ChatPromptTemplate

prompt_template = ChatPromptTemplate.from_messages([
    ("system", "تو یک دستیار هوشمند و متخصص در حوزه {domain} هستی."),
    ("user", "لطفاً درباره موضوع زیر توضیح بده: {topic}")
])

در این قطعه کد، کلیدهای domain و topic به عنوان متغیرهای ورودی قالب تعریف شده‌اند. انتخاب این کلیدها بر اساس نیاز تجاری انجام شده است تا ورودی‌های ساختاریافته‌ای از بخش‌های دیگر برنامه دریافت کنیم. تغییر نام این کلیدها نیازمند به‌روزرسانی دیکشنری ورودی در سراسر زنجیره خواهد بود.

اشتباه رایج: عدم مطابقت نام کلیدهای ورودی در دیکشنری با متغیرهای تعریف شده در PromptTemplate باعث بروز خطای زمان اجرای کلید گم‌شده (KeyError) می‌شود. نام کلیدها باید دقیقاً یکسان باشد.

برای درک بهتر نحوه کامپایل شدن این ساختار، خروجی نهایی آماده ارسال به مدل را بررسی می‌کنیم:

messages = prompt_template.invoke({
    "domain": "برنامه‌نویسی پایتون", 
    "topic": "مدیریت حافظه"
})

print(messages)

خروجی واقعی این دستور شامل لیست پیام‌های ساختاریافته زیر است:

[
    SystemMessage(content="تو یک دستیار هوشمند و متخصص در حوزه برنامه‌نویسی پایتون هستی."),
    HumanMessage(content="لطفاً درباره موضوع زیر توضیح بده: مدیریت حافظه")
]

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

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

model = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

chain = prompt_template | model | parser

response = chain.invoke({
    "domain": "هوش مصنوعی",
    "topic": "شبکه‌های عصبی کانولوشنی"
})

در این زنجیره پیشرفته، ورودی‌های دیکشنری ابتدا وارد ChatPromptTemplate می‌شوند تا پیام‌های استاندارد تولید گردند. سپس این پیام‌ها به مدل ارسال شده و در نهایت خروجی مدل توسط StrOutputParser به یک رشته متنی تمیز تبدیل می‌شود.

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

پاکسازی و ساختاردهی خروجی با Output Parsers

مدل‌های زبانی بزرگ ذاتاً متن خام تولید می‌کنند، اما برنامه‌های ما برای پردازش‌های بعدی به ساختارهای داده‌ای مشخص مثل رشته تمیز، ساختار JSON یا لیست‌های تایپ‌شده نیاز دارند. اگر برای استخراج اطلاعات از خروجی متنی مدل به سراغ عبارات با قاعده یا همان Regex بروید، با کدهایی شکننده و غیرقابل اعتماد مواجه می‌شوید که با کوچک‌ترین تغییر در لحن پاسخ مدل از کار می‌افتند.

راهکار حرفه‌ای در مهندسی نرم‌افزار، بهره‌گیری از Output Parserها به عنوان آخرین حلقه زنجیره LCEL است تا ساختار خروجی از پیش تضمین شود. بدون استفاده از این ابزارها، سیستم‌های پایین‌دستی در پردازش پاسخ‌های نامنظم مدل دچار خطای بحرانی می‌شوند. در این حالت، توسعه‌دهنده ناچار است منطق‌های پیچیده‌ای برای اعتبارسنجی بنویسد که نگهداری کد را دشوار می‌کند.

[LLM Raw Output] -> [StrOutputParser / JsonOutputParser] -> [Structured Python Object]

برای پیاده‌سازی این مکانیزم، از کلاس‌های آماده لایه پردازش خروجی استفاده می‌کنیم. در ادامه، StrOutputParser را به انتهای لوله پرامپت و مدل اضافه می‌کنیم تا خروجی از ساختار پیام به متن خالص تبدیل شود.

from langchain_core.output_parsers import StrOutputParser

parser = StrOutputParser()
chain = prompt | model | parser
response = chain.invoke({"topic": "هوش مصنوعی"})
print(type(response))

مقدار متغیر response در این سناریو یک رشته متنی تمیز است. خروجی واقعی این اجرا به شکل زیر خواهد بود:

<class 'str'>

انتخاب StrOutputParser زمانی انجام می‌شود که فقط به محتوای متنی پاسخ نیاز داریم و آبجکت‌های پیچیده مد نظر نیستند. اگر ساختار خروجی مدل تغییر کند، این پارسر بدون اعمال تغییر در نوع داده خروجی، آن را تحویل می‌دهد.

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

برای دریافت داده‌های ساختاریافته مانند دیکشنری‌های قابل پردازش در پایتون، از JsonOutputParser بهره می‌گیریم. این پارسر متن خام مدل را تجزیه کرده و آن را به یک دیکشنری استاندارد تبدیل می‌کند.

from langchain_core.output_parsers import JsonOutputParser

json_parser = JsonOutputParser()
chain_json = prompt_json | model | json_parser
result = chain_json.invoke({"topic": "پایتون"})
print(type(result))

خروجی اجرای این قطعه کد یک دیکشنری استاندارد پایتون است که مستقیماً در برنامه‌ها قابل استفاده می‌باشد:

<class 'dict'>

اگر مدل زبانی پاسخ نامعتبری تولید کند که با ساختار JSON همخوانی نداشته باشد، پارسرها خطای ساختاری پرتاب می‌کنند. برای مدیریت این خطاهای احتمالی، استفاده از بلوک‌های مدیریت خطا در سطح برنامه ضروری است. تنظیم دقیق پرامپت با ذکر مثال، احتمال بروز این خطاهای فرمتی را به حداقل می‌رساند.

برای سناریوهای سازمانی که به تایپ‌سیفتی کامل نیاز داریم، PydanticOutputParser گزینه مهندسی برتری است. این ابزار با استفاده از مدل‌های Pydantic، داده‌های خروجی را با دقت بالا اعتبارسنجی می‌کند.

استفاده از StrOutputParser: مناسب برای پاسخ‌های متنی ساده و وبلاگ‌نویسی.
استفاده از JsonOutputParser: مناسب برای تبادل داده‌های پویا بدون اعتبارسنجی سختگیرانه.
استفاده از PydanticOutputParser: مناسب برای سیستم‌های مالی و APIهایی که به ساختار دقیق داده وابسته‌اند.

در این بخش آموختیم چگونه با به‌کارگیری Output Parserها، خروجی‌های متنی مدل را به داده‌های ساختاریافته و قابل اتکا تبدیل کنیم.

موازی‌سازی اجرای وظایف با RunnableParallel

زنجیره‌های خطی ساده در پردازش‌های حجیم زبانی به‌شدت کند هستند؛ زیرا اگر بخواهیم یک ورودی را به طور همزمان برای چند هدف مختلف (مثلاً خلاصه‌سازی و ترجمه) پردازش کنیم، اجرای ترتیبی وظایف زمان زیادی از چرخه حیات درخواست را می‌گیرد.

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

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

Input -> [RunnableParallel] -> Branch A (Summarize) + Branch B (Translate) -> Combined Dict.

برای درک نحوه عملکرد این ساختار، اجازه دهید کار را با تعریف یک ساختار موازی ساده آغاز کنیم که ورودی واحدی را به دو وظیفه مجزا ارسال می‌کند. در این الگو، کلیدهای دیکشنری مشخص می‌کنند که خروجی هر بخش در چه ساختاری قرار گیرد.

from langchain_core.runnables import RunnableParallel

chain = RunnableParallel(
    summary=summarize_chain,
    translation=translate_chain
)

result = chain.invoke({"text": "داده‌های متنی ورودی"})
print(result)

در قطعه کد بالا، ورودی دریافتی به طور همزمان به دو متغیر summary و translation فرستاده می‌شود. تخصیص این وظایف به شاخه‌های مجزا کمک می‌کند تا مصرف منابع سخت‌افزاری بهینه‌سازی شود و زمان انتظار نهایی معادل طولانی‌ترین شاخه منفرد باشد، نه مجموع زمان تمام شاخه‌ها.

خروجی اجرای این ساختار موازی یک دیکشنری ترکیبی است که خروجی هر پردازش را در کلید مربوط به خود نگه می‌دارد:

{
    "summary": "خلاصه متن در یک خط...",
    "translation": "Translated text in English..."
}

مقادیر کلیدی این دیکشنری به صورت مستقیم با نام‌هایی که در زمان تعریف RunnableParallel مشخص کرده‌ایم مطابقت دارند. اگر نام کلیدها را تغییر دهیم، ساختار داده‌های خروجی در لایه‌های بعدی برنامه‌نویسی نیز متناسب با همان کلیدها تغییر خواهد کرد.

نکته کاربردی: در استفاده از RunnableParallel مراقب محدودیت‌های نرخ درخواست (Rate Limit) ارائه‌دهندگان مدل باشید؛ زیرا اجرای همزمان شاخه‌ها تعداد درخواست‌های ارسالی به API را در یک لحظه به شدت افزایش می‌دهد.

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

در این بخش آموختیم چگونه با به‌کارگیری RunnableParallel، وظایف پردازشی را به شاخه‌های همزمان تقسیم کنیم و زمان پاسخگویی سیستم را به شکل چشمگیری بهبود دهیم.

انتقال حالت و داده‌ها با RunnablePassthrough و دیکشنری‌ها

گاهی اوقات در میان یک زنجیره پردازشی، به ورودی اصلی نیاز دارید که باید بدون هیچ تغییری به مراحل بعدی منتقل شود؛ اما سیستم جریان داده شما را به ناچار دستکاری می‌کند. اگر متغیر ورودی را در اسکوپ بالاتر ذخیره کنید و به شکل گلوبال به آن دسترسی داشته باشید، با یک ضدالگوی خطرناک در برنامه‌نویسی مواجه می‌شوید که تست‌پذیری کد را نابود می‌کند. برای حل این مشکل مهندسی، از RunnablePassthrough استفاده می‌کنیم تا داده‌ها بدون دستکاری عبور کنند یا مقادیر جدید به دیکشنری جریان بدون خطر الحاق شوند.

Input Dict -> RunnablePassthrough (Keep Original) + Computed Value -> Combined Payload

برای درک بهتر رفتار این ابزار، فرض کنید در یک سیستم پرسش و پاسخ (RAG) می‌خواهیم متن سوال اولیه کاربر را در کنار اسناد بازیابی شده حفظ کنیم تا در نهایت مدل زبانی هم به سوال و هم به اسناد دسترسی داشته باشد. اگر از رویکرد سنتی استفاده کنیم، مدیریت این متغیرهای جانبی کد ما را به شدت پیچیده و مستعد خطا می‌کند.

from langchain_core.runnables import RunnablePassthrough

chain = {"context": retriever, "question": RunnablePassthrough()}

در قطعه کد بالا، RunnablePassthrough بدون اینکه تغییری در پرسش اولیه کاربر ایجاد کند، آن را به عنوان مقدار کلید question در دیکشنری نهایی قرار می‌دهد. در همین حال، کلید context نتایج جستجوی بازیاب را دریافت و پردازش می‌کند. مقدار question دقیقاً همان رشته متنی است که کاربر ارسال کرده است و هیچ دستکاری روی آن انجام نمی‌شود.

نکته کاربردی: هرگاه دیدید برای دسترسی به ورودی اولیه مجبور به تعریف متغیرهای گلوبال یا تودرتو شده‌اید، بدانید که جای خالی RunnablePassthrough در معماری جریان شما احساس می‌شود.

برای بررسی خروجی این ساختار، فرض کنید ورودی ما پرسش مربوط به ساختار داده‌ها است. خروجی نهایی این جریان یک دیکشنری ترکیبی خواهد بود که هم متن خام و هم اسناد پردازش شده را در خود جای می‌دهد.

# خروجی نهایی جریان داده
{
    "context": [Document(page_content="اسناد مرتبط با معماری...")],
    "question": "معماری جریان داده چیست؟"
}

این خروجی به مدل زبانی اجازه می‌دهد با یک دسترسی تمیز و ساختاریافته، هم به متن سوال و هم به داده‌های زمینه‌ای دسترسی داشته باشد. فیلد context داده‌های جدید تولید شده توسط سیستم است و فیلد question همان داده حفظ شده از ورودی اولیه است.

در سناریوهای پیشرفته‌تر، ممکن است بخواهیم داده‌های پویا را به جریان اضافه کنیم بدون اینکه کل ساختار دیکشنری را دستی بازنویسی نماییم. برای این منظور از متد الحاقی آن استفاده می‌کنیم.

from langchain_core.runnables import RunnablePassthrough

enhanced_chain = RunnablePassthrough.assign(
    extra_field=lambda x: len(x["question"])
)

استفاده از RunnablePassthrough.assign به ما اجازه می‌دهد فیلدهای جدیدی را بر اساس داده‌های موجود در جریان محاسبه و به دیکشنری اضافه کنیم. در اینجا تابع لامدا طول متن سوال را به عنوان یک متریک پردازشی جدید به پِی‌لود در حال گذار الحاق می‌کند. این شیوه مدیریت حالت، امنیت جریان داده را تضمین کرده و از بروز خطاهای تغییر ناپذیری جلوگیری می‌کند.

در این بخش آموختیم چگونه با به‌کارگیری RunnablePassthrough و دیکشنری‌ها، حالت و داده‌های اصلی را بدون توسل به الگوهای مخرب گلوبال در طول زنجیره حفظ و مدیریت کنیم.

مدیریت خطا، فالت تلرنس و Fallbackها در لوله‌های هوش مصنوعی

سرویس‌های هوش مصنوعی ابری مانند APIهای OpenAI یا Anthropic همواره در معرض قطعی شبکه، نوسانات سرور یا اعمال محدودیت نرخ هستند. اگر برنامه شما بدون مکانیزم بازیابی خطا طراحی شده باشد، یک خطای ساده در دریافت پاسخ مدل باعث توقف کل فرآیند و ایجاد تجربه کاربری نامطلوب می‌شود.

شیوه‌ای که یک توسعه‌دهنده تازه‌کار برای حل این مشکل انتخاب می‌کند، نوشتن بلوک‌های طولانی از try/except دور هر فراخوانی است. این کار ساختار کد را آشفته کرده و منطق اصلی برنامه را درگیر مدیریت استثنائات پیچیده می‌کند. رویکرد مهندسی این است که تاب‌آوری را مستقیماً در سطح پروتکل Runnable و به صورت ساختاریافته پیاده‌سازی کنیم تا سیستم در مواجهه با خطا به طور خودکار مسیر جایگزین را انتخاب کند.

Primary Chain -> Error Occurs -> Fallback Chain Activated -> Valid Output

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

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

primary_model = ChatOpenAI(model="gpt-4", temperature=0)
fallback_model = ChatAnthropic(model="claude-3-sonnet", temperature=0)

robust_chain = primary_model.with_fallbacks([fallback_model])
response = robust_chain.invoke("تفاوت معماری میکروسرویس و مونولیت چیست؟")
print(response.content)

در قطعه کد بالا، مدل GPT-4 به عنوان ارائه‌دهنده اصلی و مدل Claude-3 به عنوان پشتیبان پیکربندی شده‌اند. اگر سرور ارائه‌دهنده مدل اصلی به دلیل ترافیک بالا پاسخگو نباشد، لایه LCEL بلافاصله خطا را رهگیری کرده و درخواست را به مدل دوم ارسال می‌کند. این جابجایی کاملاً شفاف انجام می‌شود و لایه‌های بالایی برنامه هیچ خطایی دریافت نمی‌کنند.

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

برای مدیریت دقیق‌تر رفتار سیستم در مواجهه با خطاها، می‌توانیم پارامترهای کنترلی را تنظیم کنیم. فرض کنید می‌خواهیم تعداد تلاش مجدد روی مدل اصلی قبل از فعال‌سازی Fallback مشخص باشد تا از جابجایی زودهنگام جلوگیری کنیم.

from langchain_core.runnables import RunnableConfig

config = RunnableConfig(
    max_retries=3,
    timeout=5.0
)

resilient_chain = primary_model.with_fallbacks(
    [fallback_model],
    exceptions_to_handle=(ConnectionError, TimeoutError)
)

مقدار تنظیم‌شده برای max_retries برابر با ۳ تعیین شده است تا در صورت بروز خطاهای موقت شبکه، درخواست مجدداً تلاش شود. مقدار timeout روی ۵.۰ ثانیه تنظیم شده تا اگر پاسخگویی مدل بیشتر از این زمان طول کشید، درخواست متوقف شده و سریعاً به مسیر جایگزین هدایت شود. این تنظیمات از معلق ماندن منابع سرور و ایجاد ترافیک انباشته جلوگیری می‌کنند.

در ادامه، تغییرات رفتار سیستم را در قالب سناریوهای مختلف ارائه‌دهندگان بررسی می‌کنیم:

سناریو اول (تک ارائه‌دهنده): استفاده از gpt-3.5-turbo به عنوان اصلی و gpt-4 به عنوان پشتیبان برای مدیریت محدودیت‌های نرخ مصرف.
سناریو دوم (چند ارائه‌دهنده): ترکیب OpenAI با مدل‌های متن‌باز محلی مانند Ollama برای تضمین استقلال کامل از سرویس‌های ابری خارجی.
سناریو سوم (زنجیره کامل): اتصال یک زنجیره پردازشی شامل پرامپت و پارسر به عنوان فالت‌تلرنس یکپارچه.

from langchain_community.llms import Ollama

local_fallback = Ollama(model="llama3")
enterprise_chain = primary_model.with_fallbacks([local_fallback])

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

یکپارچه‌سازی نهایی: ساخت یک سامانه RAG تولیدی و ماجولار با LCEL

وقتی چندین کامپوننت مستقل مانند پایگاه داده وکتوری، LLM (مدل زبانی بزرگ) و قالب پرامپت را توسعه می‌دهیم، بزرگ‌ترین چالش در محیط پروداکشن، مدیریت خطوط انتقال داده و جلوگیری از پراکندگی منطق برنامه است. اگر این اجزا را با کدهای اسکریپتی و دستی به هم متصل کنیم، با اولین تغییر ساختار داده در ورودی یا خروجی، کل سیستم دچار شکستگی و خطاهای پنهان در زمان اجرا می‌شود.

برای حل این مسئله حیاتی، تمام اجزای سامانه را در قالب یک جریان یکپارچه مبتنی بر LCEL پیاده‌سازی می‌کنیم تا مدیریت وضعیت و انتقال داده‌ها به صورت خودکار و ایمن انجام شود.

در این معماری، پرسش کاربر به صورت موازی پردازش می‌شود؛ از یک سو برای جستجوی اسناد مرتبط به پایگاه داده وکتوری ارسال می‌شود و از سوی دیگر، خود پرسش بدون تغییر برای مرحله ترکیب نهایی حفظ می‌گردد.

User Query ──> [ Parallel: (Retrieve Context) + (Keep Query) ] ──> Prompt Template ──> LLM ──> Output Parser

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

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

retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}
)

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

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

setup_retrieval = RunnableParallel(
    context=retriever | format_docs,
    question=RunnablePassthrough()
)

کلاس RunnableParallel به سیستم اجازه می‌دهد عملیات جستجو و انتقال متن را همزمان انجام دهد و RunnablePassthrough تضمین می‌کند که سوال اصلی کاربر بدون دستکاری به مرحله ساخت پرامپت برسد.

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

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

prompt = ChatPromptTemplate.from_template(
    "با توجه به اطلاعات زیر:\n\n{context}\n\nبه این سوال پاسخ دهید: {question}"
)

rag_chain = setup_retrieval | prompt | llm | StrOutputParser()

response = rag_chain.invoke("معماری سیستم‌های ابری چیست؟")
print(response)

خروجی مستند این سیستم در محیط پروداکشن به شکل زیر خواهد بود که پاسخ دقیق را به همراه استناد به منابع بازیابی شده ارائه می‌دهد.

سیستم‌های ابری به معماری‌هایی اطلاق می‌شوند که منابع محاسباتی... (منابع: سند شماره ۲، سند شماره ۴)

برای ارتقای این سامانه به سطوح بالاتر، می‌توانیم زنجیره را توسعه دهیم تا پیش از تحویل پاسخ به کاربر، مرحله ارزیابی صحت (Guardrails) روی خروجی اعمال شود. در این بخش آموختیم چگونه با ترکیب دقیق ابزارهای LCEL، یک سامانه RAG مقیاس‌پذیر، پایدار و آماده برای استفاده در محصولات واقعی بسازیم.

بهینه‌سازی عملکرد، استریمینگ (Streaming) و الگوهای پیشرفته پایش

در سامانه‌های تجاری، سرعت پاسخگویی و قابلیت مانیتورینگ زنجیره‌ها برای ارائه تجربه کاربری بی‌نقص حیاتی است. وقتی کاربر منتظر پاسخ هوش مصنوعی است، هر ثانیه تأخیر می‌تواند به قیمت از دست رفتن مخاطب تمام شود.

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

به جای این انتظار فرساینده، از متد astream برای استریم کردن لحظه‌ای توکن‌ها در سراسر زنجیره LCEL استفاده می‌کنیم. این متد داده‌ها را به محض تولید توسط مدل، قطعه‌قطعه (Chunk) به لایه نمایش منتقل می‌کند تا کاربر بلافاصله شاهد پیشرفت کار باشد.

از منظر معماری توزیع‌شده، پایش دقیق جریان داده‌ها در گره‌های مختلف برای عیب‌یابی خطاهای احتمالی یک ضرورت غیرقابل انکار است. بدون ابزارهای رهگیری مناسب، متوجه نمی‌شویم که تقاضای کاربر در کدام مرحله از خط لوله با کندی یا خطا مواجه شده است.

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

Client Request ---> astream() Pipeline ---> Real-time Token Chunks ---> UI Display
                         |
                         v
                  LangSmith Tracing

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

async for chunk in chain.astream({"topic": "هوش مصنوعی"}):
    print(chunk, end="", flush=True)

در قطعه کد بالا، متد astream یک مولد آسنکرون (Async Generator) ایجاد می‌کند که به جای بازگرداندن یکجای پاسخ، توکن‌ها را به محض آماده شدن پمپ می‌کند. پارامتر end="" در تابع چاپ، از انتقال اضافی خطوط جلوگیری می‌کند و پارامتر flush=True تضمین می‌کند که بافر خروجی بلافاصله در ترمینال تخلیه شود.

نکته کاربردی: همیشه در محیط‌های پربار (High-Load) از نسخه‌های آسنکرون مانند astream به جای نسخه‌های سنکرون استفاده کنید تا از مسدود شدن ترد اصلی سرور جلوگیری شود.

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

async for chunk in structured_chain.astream({"query": "تحلیل بازار"}):
    if "result" in chunk:
        print(chunk["result"], end="", flush=True)

خروجی واقعی این اجرا در ترمینال به شکل زیر ظاهر می‌شود که نشان‌دهنده جریان مداوم اطلاعات است:

در حال بررسی داده‌های بازار... رشد تکنولوژی‌های ابری در سال جاری میلادی سرعت بیشتری به خود گرفته است.

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

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

این مطلب رو دوست داشتی؟ با دوستات به اشتراک بزار 😋