راهنمای جامع Agentic RAG: معماری، الگوهای طراحی و پیاده‌سازی عملی در ۲۰۲۶

اگه چند ماه اخیر تو حوزه هوش مصنوعی فعال بوده باشید، حتماً اسم Agentic RAG به گوشتون خورده. سیستم‌های بازیابی تقویت‌شده تولیدی (RAG) مدت‌هاست که یکی از ستون‌های اصلی توسعه اپلیکیشن‌های مبتنی بر مدل‌های زبانی بزرگ هستند. اما چیزی که الان در سال ۲۰۲۶ داریم می‌بینیم، یه تحول جدی‌ه: گذار از خط‌لوله‌های ساده بازیابی-تولید به سیستم‌های هوشمندی که خودشون استدلال می‌کنند، تصمیم می‌گیرند و حتی خطاهاشون رو اصلاح می‌کنند.

در این مقاله قراره از صفر تا صد این موضوع رو با هم بررسی کنیم.

۱. مقدمه: از RAG تا Agentic RAG

RAG چیست و چرا اهمیت دارد؟

بازیابی تقویت‌شده تولیدی یا Retrieval-Augmented Generation (RAG) رویکردی‌ه که توش مدل زبانی بزرگ به جای اینکه فقط به دانش داخلی خودش تکیه کنه، اول اطلاعات مرتبط رو از منابع خارجی بازیابی می‌کنه و بعد با کمک اون اطلاعات، پاسخ دقیق‌تر و به‌روزتری تولید می‌کنه. این رویکرد اولین بار توسط تیم تحقیقاتی Meta در سال ۲۰۲۰ معرفی شد و به مشکلات اساسی مدل‌های زبانی جواب می‌ده — از جمله توهم‌زایی (hallucination)، منسوخ بودن اطلاعات و عدم دسترسی به داده‌های اختصاصی سازمانی.

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

چرا Agentic RAG ظهور کرد؟

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

Agentic RAG یا بازیابی تقویت‌شده عامل‌محور، دقیقاً برای حل این مشکلات متولد شده. در این رویکرد، مدل زبانی بزرگ دیگه صرفاً یه تولیدکننده متن نیست بلکه نقش یک موتور استدلال (reasoning engine) رو ایفا می‌کنه. می‌تونه تصمیم بگیره کِی بازیابی کنه، از کدوم منابع استفاده کنه، کیفیت اسناد رو ارزیابی کنه و در صورت نیاز، پرسش رو بازنویسی کنه و فرآیند رو تکرار کنه.

به بیان ساده‌تر: Agentic RAG یک خط‌لوله نیست بلکه یک حلقه (loop) هست.

و اعداد هم حرف خودشون رو می‌زنن. بر اساس گزارش‌های Gartner، پرسش‌های مربوط به سیستم‌های چندعامله از سه‌ماهه اول ۲۰۲۴ تا سه‌ماهه دوم ۲۰۲۵ رشد خیره‌کننده ۱,۴۴۵ درصدی رو تجربه کردن. همچنین پیش‌بینی می‌شه تا پایان سال ۲۰۲۶، حدود ۴۰ درصد از کاربردهای سازمانی از عامل‌های هوش مصنوعی بهره ببرن — در حالی که این رقم تو سال ۲۰۲۵ کمتر از ۵ درصد بود.

۲. تفاوت RAG سنتی و Agentic RAG

RAG سنتی: خط‌لوله بازیابی-تولید

تو RAG سنتی، جریان داده کاملاً خطی و از پیش تعیین‌شده‌ست:

1. دریافت پرسش: پرسش کاربر دریافت می‌شه.
2. تبدیل به بردار: پرسش توسط مدل تعبیه (embedding model) به بردار عددی تبدیل می‌شه.
3. جستجوی شباهت: بردار پرسش با بردارهای اسناد ذخیره‌شده مقایسه و مرتبط‌ترین اسناد بازیابی می‌شن.
4. تولید پاسخ: اسناد بازیابی‌شده همراه با پرسش اصلی به عنوان زمینه (context) به مدل زبانی ارسال شده و پاسخ تولید می‌شه.

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

Agentic RAG: حلقه استدلال پویا با استفاده از ابزار

Agentic RAG رویکردی بنیادین متفاوت داره. اینجا مدل زبانی به عنوان یک عامل هوشمند عمل می‌کنه با قابلیت‌های زیر:

• تجزیه پرسش (Query Decomposition): عامل پرسش پیچیده رو به زیرپرسش‌های ساده‌تر تقسیم می‌کنه و هر کدوم رو مستقل پردازش می‌کنه.
• انتخاب پویای منابع: بر اساس ماهیت پرسش تصمیم می‌گیره از کدوم منابع (پایگاه داده برداری، وب، APIها، گراف دانش) اطلاعات رو بگیره.
• ارزیابی و درجه‌بندی: اسناد بازیابی‌شده از نظر مرتبط بودن و کیفیت ارزیابی می‌شن و اگه کافی نباشن، فرآیند بازیابی تکرار می‌شه.
• خودتصحیحی: عامل می‌تونه پاسخ خودش رو نقد کنه و در صورت شناسایی مشکل، اصلاحش کنه.
• استفاده از ابزار: به مجموعه‌ای از ابزارها دسترسی داره و بر اساس نیاز از اون‌ها استفاده می‌کنه.

خلاصه‌اش اینه: RAG سنتی سریع‌تر و ارزان‌تره ولی محدودیت‌های بیشتری داره. Agentic RAG تأخیر بیشتری رو به ازای قابلیت اطمینان و دقت بالاتر قبول می‌کنه.

۳. معماری سیستم‌های Agentic RAG

معماری یک سیستم Agentic RAG از سه لایه اصلی تشکیل شده. بیاید هر کدوم رو بررسی کنیم:

لایه اول: سیستم بازیابی (Retrieval System)

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

• پایگاه داده برداری (Vector Store): برای ذخیره و جستجوی تعبیه‌های اسناد. ابزارهایی مثل Chroma، Pinecone، Weaviate و Qdrant در این دسته قرار می‌گیرن.
• موتور جستجوی کلیدواژه‌ای: مکمل جستجوی برداری برای مواردی که تطابق دقیق کلمات مهمه (مثلاً BM25).
• پایگاه داده گراف دانش: برای ذخیره و پیمایش روابط بین موجودیت‌ها.
• ابزارهای جستجوی وب: برای دسترسی به اطلاعات به‌روز و خارج از پایگاه داده داخلی.
• رابط‌های API: برای اتصال به سیستم‌های سازمانی و منابع داده ساختاریافته.

لایه دوم: مدل تولیدی (Generation Model)

مدل زبانی بزرگ وظیفه تولید پاسخ نهایی رو بر عهده داره. ولی نکته اینجاست که تو Agentic RAG، نقش مدل فراتر از صرفاً تولید متنه. مدل باید بتونه:

• زمینه بازیابی‌شده رو درک و تحلیل کنه.
• اطلاعات از منابع مختلف رو ترکیب و سنتز کنه.
• توانایی فراخوانی ابزار (function calling) داشته باشه.
• خروجی ساختاریافته تولید کنه (برای تعامل با سایر اجزای سیستم).

لایه سوم: لایه عامل (Agent Layer)

این لایه عملاً مغز متفکر سیستمه و هماهنگی بین لایه‌های بازیابی و تولید رو مدیریت می‌کنه. وظایفش شامل موارد زیره:

• تجزیه پرسش (Query Decomposition): تبدیل پرسش‌های پیچیده به زیرپرسش‌های قابل مدیریت.
• مسیریابی (Routing): تصمیم‌گیری درباره اینکه هر زیرپرسش باید به کدوم منبع بره.
• ارزیابی (Grading): بررسی کیفیت و مرتبط بودن اسناد بازیابی‌شده.
• بازنویسی پرسش (Query Rewriting): اصلاح پرسش وقتی نتایج کافی نباشن.
• خودتصحیحی (Self-Correction): ارزیابی پاسخ نهایی و اصلاح در صورت نیاز.
• مدیریت حافظه (Memory Management): نگهداری تاریخچه تعاملات و نتایج میانی.

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

۴. الگوهای طراحی کلیدی

خب، بریم سراغ الگوهای طراحی. در طراحی سیستم‌های Agentic RAG چند الگوی اصلی شناسایی شده که هر کدوم برای سناریوهای خاصی بهینه‌ان:

۴.۱. Corrective RAG (CRAG) — بازیابی تصحیحی

تو الگوی CRAG، سیستم بعد از بازیابی اسناد، یه مرحله ارزیابی کیفیت (quality grading) انجام می‌ده. عامل هر سند بازیابی‌شده رو بر اساس مرتبط بودن، صحت و کامل بودن درجه‌بندی می‌کنه. اگه کیفیت کافی نباشه، یکی از این کارها رو انجام می‌ده:

• بازنویسی پرسش و تکرار بازیابی از پایگاه داده برداری
• جستجو در منابع جایگزین (مثلاً جستجوی وب)
• درخواست اطلاعات تکمیلی از کاربر

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

۴.۲. Adaptive RAG — بازیابی تطبیقی

الگوی Adaptive RAG از یه مسیریاب هوشمند استفاده می‌کنه که بر اساس تحلیل پرسش ورودی، بهترین استراتژی بازیابی رو انتخاب می‌کنه. این مسیریاب ممکنه پرسش رو به یکی از مسیرهای زیر هدایت کنه:

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

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

۴.۳. Self-RAG — بازیابی خودبازتابی

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

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

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

۴.۴. Multi-Agent RAG — بازیابی چندعامله

تو این الگو به جای یک عامل واحد، چند عامل تخصصی به صورت هماهنگ کار می‌کنن:

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

الگوی چندعامله مخصوصاً برای سازمان‌های بزرگ با منابع داده متنوع مناسبه و مقیاس‌پذیری بالاتری نسبت به بقیه الگوها داره.

۵. پیاده‌سازی عملی با LangGraph

خب، حالا بریم سراغ بخش عملی قضیه. تو این بخش یه پیاده‌سازی کامل از Agentic RAG با فریم‌ورک LangGraph ارائه می‌دیم. LangGraph یکی از قوی‌ترین ابزارها برای ساخت جریان‌های کاری عامل‌محوره که امکان تعریف گراف‌های حالت (state graphs) با گره‌ها، یال‌ها و منطق شرطی رو فراهم می‌کنه.

۵.۱. نصب وابستگی‌ها

اول از همه، کتابخانه‌های مورد نیاز رو نصب می‌کنیم:

pip install langchain langgraph langchain-openai langchain-community \
    chromadb tiktoken langchain-chroma beautifulsoup4

۵.۲. بارگذاری و آماده‌سازی اسناد

تو مرحله اول، اسناد رو بارگذاری کرده و به قطعات کوچک‌تر (chunks) تقسیم می‌کنیم:

import os
from langchain_community.document_loaders import WebBaseLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

os.environ["OPENAI_API_KEY"] = "your-api-key-here"

# Load documents from web sources
urls = [
    "https://docs.example.com/agentic-rag-overview",
    "https://docs.example.com/langgraph-tutorial",
    "https://docs.example.com/vector-databases-guide",
]

docs = []
for url in urls:
    loader = WebBaseLoader(url)
    docs.extend(loader.load())

# Split documents into chunks
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " ", ""],
)
doc_chunks = text_splitter.split_documents(docs)

print(f"Total chunks created: {len(doc_chunks)}")

# Create vector store with Chroma
embedding_model = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    documents=doc_chunks,
    embedding=embedding_model,
    collection_name="agentic-rag-docs",
    persist_directory="./chroma_db",
)

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

۵.۳. تعریف حالت و ابزارها

حالا ساختار حالت (state) گراف و ابزار بازیابی رو تعریف می‌کنیم:

from typing import TypedDict, List, Annotated
from langchain_core.documents import Document
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
from langchain_core.tools import tool


class AgenticRAGState(TypedDict):
    """State for the Agentic RAG workflow."""
    messages: Annotated[List[BaseMessage], add_messages]
    question: str
    documents: List[Document]
    generation: str
    retry_count: int
    web_search_needed: bool


@tool
def retrieve_documents(query: str) -> List[Document]:
    """Retrieve relevant documents from the vector store based on the query."""
    docs = retriever.invoke(query)
    return docs


@tool
def web_search(query: str) -> str:
    """Search the web for additional information when local documents are insufficient."""
    from langchain_community.tools.tavily_search import TavilySearchResults
    search_tool = TavilySearchResults(max_results=3)
    results = search_tool.invoke({"query": query})
    return "\n\n".join([r["content"] for r in results])

۵.۴. تعریف گره‌های گراف

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

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

llm = ChatOpenAI(model="gpt-4o", temperature=0)


def retrieve_node(state: AgenticRAGState) -> dict:
    """Retrieve documents relevant to the user's question."""
    print("--- RETRIEVE NODE ---")
    question = state["question"]
    documents = retrieve_documents.invoke(question)
    return {
        "documents": documents,
        "question": question,
    }


def grade_documents_node(state: AgenticRAGState) -> dict:
    """Grade retrieved documents for relevance to the question."""
    print("--- GRADE DOCUMENTS NODE ---")
    question = state["question"]
    documents = state["documents"]

    grading_prompt = ChatPromptTemplate.from_messages([
        ("system", """You are a document relevance grader. Given a user question
        and a retrieved document, determine if the document is relevant to answering
        the question. Respond with a JSON object: {{"relevant": "yes"}} or
        {{"relevant": "no"}}."""),
        ("human", "Question: {question}\n\nDocument:\n{document}"),
    ])

    grading_chain = grading_prompt | llm | JsonOutputParser()

    relevant_docs = []
    web_search_needed = False

    for doc in documents:
        result = grading_chain.invoke({
            "question": question,
            "document": doc.page_content,
        })
        if result.get("relevant") == "yes":
            relevant_docs.append(doc)

    # If less than 2 relevant documents found, flag for web search
    if len(relevant_docs) < 2:
        web_search_needed = True
        print(f"Only {len(relevant_docs)} relevant docs found. Web search needed.")
    else:
        print(f"Found {len(relevant_docs)} relevant documents.")

    return {
        "documents": relevant_docs,
        "web_search_needed": web_search_needed,
    }


def rewrite_query_node(state: AgenticRAGState) -> dict:
    """Rewrite the query to improve retrieval results."""
    print("--- REWRITE QUERY NODE ---")
    question = state["question"]

    rewrite_prompt = ChatPromptTemplate.from_messages([
        ("system", """You are a query rewriting expert. Given the original
        question that did not yield good retrieval results, rewrite it to be
        more specific and likely to match relevant documents. Return only
        the rewritten question."""),
        ("human", "Original question: {question}\n\nRewritten question:"),
    ])

    rewrite_chain = rewrite_prompt | llm | StrOutputParser()
    rewritten = rewrite_chain.invoke({"question": question})
    print(f"Rewritten query: {rewritten}")

    return {
        "question": rewritten,
        "retry_count": state.get("retry_count", 0) + 1,
    }


def web_search_node(state: AgenticRAGState) -> dict:
    """Perform web search to supplement retrieved documents."""
    print("--- WEB SEARCH NODE ---")
    question = state["question"]
    documents = state.get("documents", [])

    search_results = web_search.invoke(question)

    web_doc = Document(
        page_content=search_results,
        metadata={"source": "web_search"},
    )
    documents.append(web_doc)

    return {
        "documents": documents,
        "web_search_needed": False,
    }


def generate_node(state: AgenticRAGState) -> dict:
    """Generate a final answer using retrieved documents as context."""
    print("--- GENERATE NODE ---")
    question = state["question"]
    documents = state["documents"]

    context = "\n\n---\n\n".join([doc.page_content for doc in documents])

    generate_prompt = ChatPromptTemplate.from_messages([
        ("system", """You are a helpful assistant answering questions based on
        the provided context. Use the context to give accurate and comprehensive
        answers. If the context doesn't contain enough information, clearly state
        what you know and what is uncertain.

        Context:
        {context}"""),
        ("human", "{question}"),
    ])

    generate_chain = generate_prompt | llm | StrOutputParser()
    generation = generate_chain.invoke({
        "context": context,
        "question": question,
    })

    return {"generation": generation}


def hallucination_check_node(state: AgenticRAGState) -> dict:
    """Check if the generated answer is grounded in the retrieved documents."""
    print("--- HALLUCINATION CHECK NODE ---")
    documents = state["documents"]
    generation = state["generation"]

    context = "\n\n".join([doc.page_content for doc in documents])

    check_prompt = ChatPromptTemplate.from_messages([
        ("system", """You are a hallucination detector. Given a set of source
        documents and a generated answer, determine if the answer is grounded
        in and supported by the documents. Respond with JSON:
        {{"grounded": "yes"}} or {{"grounded": "no"}}."""),
        ("human", "Documents:\n{context}\n\nGenerated Answer:\n{generation}"),
    ])

    check_chain = check_prompt | llm | JsonOutputParser()
    result = check_chain.invoke({
        "context": context,
        "generation": generation,
    })

    return {"hallucination_result": result.get("grounded", "no")}

۵.۵. ساخت گراف حالت با مسیرهای شرطی

حالا همه اجزا رو کنار هم می‌ذاریم و گراف نهایی رو با یال‌های شرطی (conditional edges) می‌سازیم. اینجا دقیقاً جاییه که جادوی Agentic RAG اتفاق می‌افته:

from langgraph.graph import StateGraph, END, START


def route_after_grading(state: AgenticRAGState) -> str:
    """Route based on document grading results."""
    retry_count = state.get("retry_count", 0)
    web_search_needed = state.get("web_search_needed", False)

    if web_search_needed and retry_count < 2:
        return "rewrite_query"
    elif web_search_needed and retry_count >= 2:
        return "web_search"
    else:
        return "generate"


def route_after_hallucination_check(state: AgenticRAGState) -> str:
    """Route based on hallucination check results."""
    if state.get("hallucination_result") == "yes":
        return "end"
    else:
        retry_count = state.get("retry_count", 0)
        if retry_count < 3:
            return "rewrite_query"
        else:
            return "end"  # Return best effort after max retries


# Build the state graph
workflow = StateGraph(AgenticRAGState)

# Add nodes
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("grade_documents", grade_documents_node)
workflow.add_node("rewrite_query", rewrite_query_node)
workflow.add_node("web_search", web_search_node)
workflow.add_node("generate", generate_node)
workflow.add_node("hallucination_check", hallucination_check_node)

# Add edges
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "grade_documents")

# Conditional routing after grading
workflow.add_conditional_edges(
    "grade_documents",
    route_after_grading,
    {
        "rewrite_query": "rewrite_query",
        "web_search": "web_search",
        "generate": "generate",
    },
)

# After rewriting, go back to retrieve
workflow.add_edge("rewrite_query", "retrieve")

# After web search, go to generate
workflow.add_edge("web_search", "generate")

# After generation, check for hallucinations
workflow.add_edge("generate", "hallucination_check")

# Conditional routing after hallucination check
workflow.add_conditional_edges(
    "hallucination_check",
    route_after_hallucination_check,
    {
        "end": END,
        "rewrite_query": "rewrite_query",
    },
)

# Compile the graph
app = workflow.compile()

print("Agentic RAG workflow compiled successfully!")

۵.۶. اجرای سیستم

و در نهایت، سیستم رو اجرا می‌کنیم و نتایج رو می‌بینیم:

def run_agentic_rag(question: str) -> str:
    """Run the Agentic RAG workflow and return the generated answer."""
    print(f"\n{'='*60}")
    print(f"Question: {question}")
    print(f"{'='*60}\n")

    initial_state = {
        "question": question,
        "documents": [],
        "generation": "",
        "retry_count": 0,
        "web_search_needed": False,
        "messages": [],
    }

    # Stream the execution to see each step
    for step in app.stream(initial_state):
        for node_name, node_output in step.items():
            print(f"\n>> Completed: {node_name}")
            if "generation" in node_output and node_output["generation"]:
                print(f"\nGenerated Answer:\n{node_output['generation'][:500]}...")

    # Get final state
    final_state = app.invoke(initial_state)
    return final_state["generation"]


# Example usage
if __name__ == "__main__":
    questions = [
        "What are the key differences between Agentic RAG and traditional RAG?",
        "How does LangGraph handle state management in agent workflows?",
        "What are the best practices for production RAG systems in 2026?",
    ]

    for q in questions:
        answer = run_agentic_rag(q)
        print(f"\n{'='*60}")
        print(f"FINAL ANSWER:\n{answer}")
        print(f"{'='*60}\n")

این پیاده‌سازی جریان کامل یه سیستم Agentic RAG رو نشون می‌ده: از بازیابی اولیه تا درجه‌بندی اسناد، بازنویسی پرسش، جستجوی وب و بررسی توهم‌زایی. گراف حالت LangGraph تضمین می‌کنه که هر گام درست مدیریت بشه و مسیرهای شرطی تصمیم‌گیری پویا رو تو هر مرحله ممکن می‌کنن.

۶. تکنیک‌های پیشرفته بازیابی

کیفیت بازیابی مستقیماً روی عملکرد کلی سیستم Agentic RAG تأثیر می‌ذاره. بیاید چند تکنیک پیشرفته رو بررسی کنیم که می‌تونن دقت بازیابی رو به شکل محسوسی بهبود بدن.

۶.۱. جستجوی ترکیبی (Hybrid Search)

جستجوی ترکیبی، ترکیبی از جستجوی برداری (معنایی) و جستجوی کلیدواژه‌ای (BM25) هست. این رویکرد نقاط قوت هر دو روش رو با هم ادغام می‌کنه:

• جستجوی برداری: تو درک مفاهیم مشابه و مترادف‌ها قویه. مثلاً «خودرو» و «اتومبیل» رو معنایی نزدیک تشخیص می‌ده.
• جستجوی BM25: تو تطابق دقیق اصطلاحات فنی، اسامی خاص و شماره‌ها بهتر عمل می‌کنه.

در عمل، نتایج هر دو روش با الگوریتم‌هایی مثل Reciprocal Rank Fusion (RRF) ادغام می‌شن. به این ترتیب اسنادی که تو هر دو روش رتبه بالایی گرفتن، اولویت بیشتری پیدا می‌کنن.

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

# Create BM25 retriever from the same documents
bm25_retriever = BM25Retriever.from_documents(
    doc_chunks,
    k=5,
)

# Create ensemble (hybrid) retriever
hybrid_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, retriever],
    weights=[0.4, 0.6],  # 40% BM25, 60% vector search
)

۶.۲. HyDE — تعبیه اسناد فرضی

Hypothetical Document Embeddings (HyDE) یه تکنیک خلاقانه‌ست. ایده‌اش اینه که به جای تبدیل مستقیم پرسش به بردار، اول مدل زبانی یه «پاسخ فرضی» به پرسش تولید می‌کنه و بعد اون پاسخ فرضی به بردار تبدیل شده و برای جستجو استفاده می‌شه.

منطقش ساده ولی هوشمندانه‌ست: بردار یه پاسخ فرضی از نظر فضای معنایی به اسناد واقعی نزدیک‌تره تا بردار یه سوال. چرا؟ چون اسناد ذخیره‌شده معمولاً جملات توصیفی هستن نه سوال.

from langchain.chains import HypotheticalDocumentEmbedder

hyde_embeddings = HypotheticalDocumentEmbedder.from_llm(
    llm=llm,
    base_embeddings=embedding_model,
    prompt_key="web_search",
)

# Use HyDE embeddings for retrieval
hyde_vectorstore = Chroma.from_documents(
    documents=doc_chunks,
    embedding=hyde_embeddings,
    collection_name="hyde-docs",
)

۶.۳. بازرتبه‌بندی با Cross-Encoder

بازرتبه‌بندی (reranking) با مدل‌های Cross-Encoder یکی از مؤثرترین تکنیک‌ها برای بهبود دقت بازیابیه. ایده‌اش اینه: اول تعداد زیادی سند بازیابی می‌شه (مثلاً ۲۰ تا)، بعد یه مدل Cross-Encoder هر جفت (پرسش، سند) رو مستقیماً ارزیابی کرده و امتیاز دقیق‌تری می‌ده. در نهایت بهترین‌ها (مثلاً ۵ تا) انتخاب می‌شن.

فرقش با Bi-Encoder (که تو جستجوی برداری استفاده می‌شه) اینه که Cross-Encoder پرسش و سند رو همزمان پردازش می‌کنه و تعاملات بینشون رو مدل می‌کنه. این تعامل مستقیم دقت بسیار بالاتری می‌ده، هرچند سرعتش کمتره.

from langchain.retrievers import ContextualCompressionRetriever
from langchain_community.cross_encoders import HuggingFaceCrossEncoder
from langchain.retrievers.document_compressors import CrossEncoderReranker

# Initialize cross-encoder model
cross_encoder = HuggingFaceCrossEncoder(model_name="cross-encoder/ms-marco-MiniLM-L-6-v2")
compressor = CrossEncoderReranker(model=cross_encoder, top_n=5)

# Create reranking retriever
reranking_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=retriever,  # First retrieve 20 docs, then rerank to top 5
)

۶.۴. استخراج موجودیت و ادغام گراف دانش

یکی از پیشرفته‌ترین تکنیک‌ها ترکیب بازیابی برداری با گراف دانش (Knowledge Graph) هست. روند کار به این شکله:

1. اول موجودیت‌ها (entities) و روابط بین‌شون از اسناد استخراج می‌شن.
2. این اطلاعات تو یه پایگاه داده گراف (مثل Neo4j) ذخیره می‌شن.
3. موقع بازیابی، عامل هم از جستجوی برداری و هم از پیمایش گراف استفاده می‌کنه.
4. اطلاعات ساختاریافته از گراف با اطلاعات غیرساختاریافته از اسناد ترکیب می‌شن.

این ترکیب مخصوصاً برای سوالاتی که نیاز به درک روابط بین مفاهیم دارن عالیه. مثلاً سوالی مثل «کدوم محصولات شرکت X از فناوری Y استفاده می‌کنن و به کدوم بازارها صادر می‌شن؟» با جستجوی برداری خیلی سخته ولی با گراف دانش راحت قابل پاسخه.

۷. چالش‌ها و بهترین شیوه‌ها در محیط تولید

انتقال یه سیستم Agentic RAG از محیط آزمایشی به محیط تولید (production) چالش‌های خاص خودش رو داره. بیاید مهم‌ترین‌هاشون رو بررسی کنیم.

۷.۱. مدیریت تأخیر (Latency Management)

این احتمالاً بزرگ‌ترین چالش Agentic RAG هست. هر حلقه تکراری شامل چندین فراخوانی مدل زبانی و بازیابیه و هر کدوم زمان خودش رو می‌گیره. چند راهکار عملی:

• اجرای موازی: زیرپرسش‌های مستقل رو موازی پردازش کنید. LangGraph از اجرای موازی گره‌های مستقل پشتیبانی می‌کنه.
• حافظه نهان (Caching): نتایج بازیابی و پاسخ‌های تکراری رو کش کنید. برای پرسش‌های مشابه از کش معنایی (semantic cache) استفاده کنید.
• پخش جریانی (Streaming): پاسخ رو تدریجی نشون بدید تا زمان انتظار محسوس کمتر بشه.
• مدل‌های کوچک‌تر برای وظایف ساده: مراحلی مثل درجه‌بندی اسناد رو با مدل‌های سبک‌تر انجام بدید. مثلاً GPT-4o-mini برای مرحله درجه‌بندی کافیه.
• محدودیت تکرار: حداکثر تعداد تکرارها رو مشخص کنید (مثلاً ۳ بار بازنویسی پرسش) تا از حلقه‌های بی‌پایان جلوگیری بشه.

۷.۲. بهینه‌سازی هزینه (Cost Optimization)

هر فراخوانی مدل زبانی هزینه داره و تو Agentic RAG تعداد فراخوانی‌ها خیلی بیشتر از RAG سنتیه. چند استراتژی مؤثر:

• سلسله‌مراتب مدل‌ها: از مدل‌های قوی‌تر (و گران‌تر) فقط برای تولید نهایی استفاده کنید. مراحل درجه‌بندی و مسیریابی رو با مدل‌های ارزان‌تر انجام بدید.
• فشرده‌سازی زمینه: قبل از ارسال اسناد به مدل، محتوای نامرتبط رو حذف و اسناد رو خلاصه کنید.
• خروج زودهنگام: اگه اسناد اولیه کیفیت بالایی دارن، مستقیماً برید سراغ تولید — بدون حلقه‌های اضافی.
• دسته‌بندی درخواست‌ها (Batching): درخواست‌های مشابه رو گروه‌بندی کنید و دسته‌ای پردازش کنید.

۷.۳. پایش مصرف توکن (Token Usage Monitoring)

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

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

ابزارهایی مثل LangSmith، Weights & Biases و Phoenix (Arize) برای این کار خیلی مفیدن و دید جامعی از عملکرد و هزینه سیستم می‌دن.

import tiktoken
from langchain_community.callbacks import get_openai_callback


def run_with_monitoring(question: str):
    """Run the Agentic RAG workflow with token usage monitoring."""
    with get_openai_callback() as cb:
        result = app.invoke({
            "question": question,
            "documents": [],
            "generation": "",
            "retry_count": 0,
            "web_search_needed": False,
            "messages": [],
        })

        print(f"\n--- Token Usage Report ---")
        print(f"Total Tokens: {cb.total_tokens:,}")
        print(f"Prompt Tokens: {cb.prompt_tokens:,}")
        print(f"Completion Tokens: {cb.completion_tokens:,}")
        print(f"Total Cost: ${cb.total_cost:.4f}")
        print(f"Successful Requests: {cb.successful_requests}")

    return result["generation"]

۷.۴. حفاظ‌ها و ایمنی (Guardrails and Safety)

سیستم‌های Agentic RAG چون استقلال بیشتری تو تصمیم‌گیری دارن، به حفاظ‌های قوی‌تری هم نیاز دارن. مهم‌ترین جنبه‌های ایمنی:

• اعتبارسنجی ورودی: فیلتر کردن پرسش‌های مخرب قبل از ورود به سیستم. این شامل شناسایی حملات تزریق پرامپت (prompt injection) هم می‌شه.
• محدودسازی دسترسی: کنترل دقیق اینکه عامل به چه منابع و ابزارهایی دسترسی داره. اصل حداقل دسترسی (least privilege) رو رعایت کنید.
• اعتبارسنجی خروجی: بررسی پاسخ نهایی از نظر صحت و عدم افشای اطلاعات حساس.
• ثبت وقایع (Logging): ثبت کامل تمام تصمیمات عامل برای بازبینی و عیب‌یابی.
• مکانیزم خاموشی اضطراری: قابلیت متوقف کردن فوری عامل در صورت رفتار غیرعادی.
• نظارت انسانی: تو سناریوهای حساس، تصمیمات مهم باید قبل از اجرا توسط انسان تأیید بشن (human-in-the-loop).

from langchain_core.runnables import RunnableLambda


def input_guardrail(state: AgenticRAGState) -> AgenticRAGState:
    """Apply input guardrails before processing the question."""
    question = state["question"]

    # Check for prompt injection patterns
    injection_patterns = [
        "ignore previous instructions",
        "system prompt",
        "you are now",
        "disregard all",
    ]

    question_lower = question.lower()
    for pattern in injection_patterns:
        if pattern in question_lower:
            raise ValueError(f"Potential prompt injection detected: {pattern}")

    # Check question length
    if len(question) > 2000:
        raise ValueError("Question exceeds maximum length of 2000 characters")

    return state


def output_guardrail(state: AgenticRAGState) -> AgenticRAGState:
    """Apply output guardrails before returning the response."""
    generation = state.get("generation", "")

    # Check for sensitive information patterns (example)
    sensitive_patterns = [
        r"\b\d{3}-\d{2}-\d{4}\b",       # SSN pattern
        r"\b\d{16}\b",                    # Credit card pattern
        r"(?i)password\s*[:=]\s*\S+",     # Password leakage
    ]

    import re
    for pattern in sensitive_patterns:
        if re.search(pattern, generation):
            state["generation"] = (
                "The response was filtered due to potentially "
                "sensitive information. Please contact support."
            )
            break

    return state

۸. آینده Agentic RAG

چشم‌انداز آینده Agentic RAG واقعاً هیجان‌انگیزه. بیاید مهم‌ترین روندها رو بررسی کنیم.

۸.۱. ادغام پروتکل MCP

پروتکل Model Context Protocol (MCP) که توسط Anthropic معرفی شده، عملاً به استاندارد واقعی (de facto standard) برای اتصال مدل‌های زبانی به منابع داده و ابزارهای خارجی تبدیل شده. این پروتکل توسط غول‌هایی مثل OpenAI، Google DeepMind و Microsoft هم پذیرفته شده.

MCP یه رابط استاندارد و یکنواخت برای اتصال عامل‌ها به انواع منابع داده فراهم می‌کنه. یعنی یه عامل Agentic RAG می‌تونه بدون نیاز به کد سفارشی برای هر منبع، به صورت پویا به هر سرور MCP وصل بشه و از ابزارها و داده‌هاش استفاده کنه. تأثیرش بر Agentic RAG عمیقه:

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

۸.۲. پذیرش سازمانی

طبق پیش‌بینی‌های Gartner، تا پایان ۲۰۲۶، ۴۰ درصد از کاربردهای سازمانی از عامل‌های هوش مصنوعی تعبیه‌شده بهره می‌برن. این رقم در مقایسه با کمتر از ۵ درصد تو سال ۲۰۲۵ رشد چشمگیریه. دلایل اصلی پشت این پذیرش گسترده:

• بلوغ ابزارها: فریم‌ورک‌هایی مثل LangGraph، CrewAI، AutoGen و Semantic Kernel به بلوغ کافی رسیدن و پیاده‌سازی رو ساده‌تر کردن.
• کاهش هزینه مدل‌ها: هزینه فراخوانی مدل‌های زبانی مدام کمتر شده و مدل‌های کوچک‌تر ولی توانمند، هزینه‌های عملیاتی رو خیلی پایین آوردن.
• اثبات ارزش تجاری: مطالعات موردی متعدد نشون دادن که این سیستم‌ها بهره‌وری قابل توجهی تو حوزه‌هایی مثل پشتیبانی مشتری، تحقیقات حقوقی و مدیریت دانش ایجاد می‌کنن.
• بهبود قابلیت اطمینان: الگوهای خودتصحیحی و حفاظ‌های ایمنی، اعتماد سازمان‌ها رو بالا بردن.

۸.۳. همگرایی پروتکل‌ها

یکی از روندهای مهم در ۲۰۲۶، حرکت به سمت همگرایی پروتکل‌های مختلف ارتباطی عامل‌هاست. سه پروتکل اصلی:

• MCP (Model Context Protocol): پروتکل اتصال مدل به ابزار و داده، توسعه‌یافته توسط Anthropic.
• A2A (Agent-to-Agent Protocol): پروتکل ارتباط مستقیم بین عامل‌ها، توسعه‌یافته توسط Google. هدفش فعال‌سازی همکاری بین عامل‌های مستقل از فریم‌ورک‌ها و فروشندگان مختلفه.
• AGNTCY: چارچوبی برای استانداردسازی مشاهده‌پذیری (observability) و مدیریت عامل‌ها تو محیط‌های تولیدی.

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

۸.۴. روندهای فناورانه آینده

علاوه بر استانداردسازی پروتکل‌ها، چند روند فناورانه دیگه هم آینده Agentic RAG رو شکل می‌دن:

• بازیابی چندوجهی (Multimodal Retrieval): سیستم‌های RAG آینده فقط به متن محدود نخواهند بود — تصاویر، نمودارها، ویدیوها و فایل‌های صوتی هم بازیابی و تحلیل خواهند شد. مدل‌هایی مثل GPT-4o و Claude این قابلیت رو فراهم کردن.
• حافظه بلندمدت عامل (Long-term Agent Memory): عامل‌ها به حافظه بلندمدت مجهز می‌شن که تجربیات و الگوهای موفق بازیابی رو ذخیره و تو تعاملات بعدی استفاده می‌کنن.
• بهینه‌سازی خودکار (Auto-Optimization): سیستم‌ها قادر خواهند بود پارامترهای خودشون رو (مثل تعداد اسناد، آستانه مرتبط بودن و انتخاب مدل) بر اساس بازخورد کاربران خودکار بهینه کنن.
• ادغام با سیستم‌های استدلال: ترکیب Agentic RAG با فرآیندهای استدلال زنجیره‌ای (chain-of-thought) و درخت فکر (tree-of-thought) برای حل مسائل پیچیده‌تر.
• GraphRAG مقیاس‌پذیر: پیشرفت در ساخت و نگهداری خودکار گراف‌های دانش در مقیاس سازمانی.

۹. نتیجه‌گیری

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

خلاصه نکات کلیدی این مقاله:

• Agentic RAG یک خط‌لوله نیست بلکه یک حلقه‌ست — مدل زبانی به عنوان موتور استدلال عمل می‌کنه و توانایی تصمیم‌گیری پویا، خودتصحیحی و استفاده از ابزار رو داره.
• الگوهای طراحی متنوعی از جمله CRAG، Adaptive RAG، Self-RAG و Multi-Agent RAG وجود دارن. انتخاب درست الگو به نیازهای کاربردی بستگی داره.
• پیاده‌سازی عملی با LangGraph نسبتاً سرراسته. گراف‌های حالت با گره‌ها و یال‌های شرطی مدل طبیعی و قدرتمندی برای جریان‌های کاری عامل‌محور ارائه می‌دن.
• تکنیک‌های پیشرفته بازیابی مثل جستجوی ترکیبی، HyDE و Cross-Encoder می‌تونن کیفیت بازیابی رو به شکل چشمگیری بهبود بدن.
• چالش‌های محیط تولید شامل مدیریت تأخیر، بهینه‌سازی هزینه و ایمنی، نیاز به توجه ویژه دارن.
• آینده این حوزه با پروتکل‌هایی مثل MCP و A2A و بازیابی چندوجهی خیلی امیدوارکننده‌ست.

با رشد ۱,۴۴۵ درصدی علاقه به سیستم‌های چندعامله و پیش‌بینی پذیرش ۴۰ درصدی تو سازمان‌ها تا پایان ۲۰۲۶، واضحه که Agentic RAG فقط یه ترند زودگذر نیست — بلکه تغییری ساختاری تو نحوه ساخت اپلیکیشن‌های هوش مصنوعیه.

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