Agentic RAG: Let AI Retrieve Autonomously

From Passive RAG to Active Agent

The traditional RAG works in a passive mode: user asks a question → retrieve documents → generate an answer. This mode performs well in simple Q&A scenarios, but when faced with complex problems that require multi-step reasoning, cross-document synthesis, and dynamic decision-making, passive RAG falls short. For example, if a user asks "Compare the patent portfolios of Company A and Company B in the AI field," passive RAG might only retrieve partial information about Company A or Company B, and cannot automatically conduct multiple rounds of retrieval and comparative analysis. Agentic RAG is designed to solve such complex retrieval scenarios.

The core idea of Agentic RAG is to introduce the autonomous decision-making capability of an Agent into the RAG system. Instead of passively executing a single retrieval, the Agent acts like a human researcher: analyze the problem → formulate a retrieval plan → execute retrieval → evaluate retrieval results → decide whether supplementary retrieval is needed → integrate all information → generate the final answer. This loop can be repeated multiple times until the Agent believes it has gathered enough information to answer the user's question.

Core Architecture of Agentic RAG

The Agentic RAG architecture consists of four core components: Planner is responsible for analyzing the user's question and formulating a retrieval strategy—breaking down complex questions into multiple sub-questions and deciding what type of information each sub-question needs to retrieve. Retriever executes the actual retrieval operations, supporting multiple retrieval methods: vector search, keyword search, structured queries, etc. Evaluator evaluates the quality of retrieval results—whether the information is sufficient to answer the user's question, whether the retrieval strategy needs adjustment, and whether supplementary retrieval is needed. Synthesizer integrates all retrieved information into a coherent answer.

from openai import OpenAI
import json

client = OpenAI(api_key="your-deepseek-api-key", base_url="https://api.deepseek.com")

class AgenticRAG:
    def __init__(self, vector_db, max_iterations=5):
        self.vector_db = vector_db
        self.max_iterations = max_iterations
        self.search_history = []

    def plan(self, question):
        """Planner: Analyze the question and formulate a retrieval plan"""
        prompt = f"""You are an expert in retrieval strategy planning. Analyze the following question and formulate a retrieval plan.

User question: {question}

Please output the retrieval plan in JSON format:
{{
  "sub_questions": [
    {{"query": "search terms for sub-question 1", "source": "knowledge_base/web/internal_docs"}},
    {{"query": "search terms for sub-question 2", "source": "knowledge_base/web/internal_docs"}}
  ],
  "reasoning": "Why it is divided this way"
}}"""
        response = client.chat.completions.create(
            model="deepseek-chat",
            messages=[{"role":"user","content":prompt}],
            temperature=0.2
        )
        return json.loads(response.choices[0].message.content)

    def retrieve(self, query, source="knowledge_base"):
        """Retriever: Execute a single retrieval"""
        # In production, route to different sources
        results = self.vector_db.search(query, top_k=5)
        return [r["content"] for r in results]

    def evaluate(self, question, collected_info):
        """Evaluator: Determine if information is sufficient"""
        prompt = f"""Evaluate whether the currently collected information is sufficient to answer the user's question.

User question: {question}
Collected information:
{json.dumps(collected_info, ensure_ascii=False)}

JSON output:
{{
  "is_sufficient": true/false,
  "missing_info": ["what additional information is needed"],
  "next_queries": ["suggested next search terms"],
  "confidence": 0.0-1.0
}}"""
        response = client.chat.completions.create(
            model="deepseek-chat",
            messages=[{"role":"user","content":prompt}],
            temperature=0.1
        )
        return json.loads(response.choices[0].message.content)

    def synthesize(self, question, all_info):
        """Synthesizer: Integrate all information to generate an answer"""
        prompt = f"""Based on the following information, answer the user's question.

User question: {question}

Collected information:
{json.dumps(all_info, ensure_ascii=False, indent=2)}

Requirements:
1. Provide a complete answer by synthesizing all information
2. Indicate the source of information (which piece of information came from which retrieval)
3. If there are contradictions between information, point them out and provide analysis
4. If certain aspects have insufficient information, clearly state that"""
        response = client.chat.completions.create(
            model="deepseek-chat",
            messages=[{"role":"user","content":prompt}],
            temperature=0.3
        )
        return response.choices[0].message.content

    def run(self, question):
        print(f"User question: {question}\n")
        plan = self.plan(question)
        print(f"Retrieval plan: {len(plan['sub_questions'])} sub-questions")

        all_info = {}
        for i, sq in enumerate(plan["sub_questions"]):
            results =