How to Implement Dynamic Query Decomposition in Agentic RAG for Complex Enterprise Search

Standard RAG systems fall apart the moment a user asks a complex enterprise question. If someone asks, "What were our main operational risks in Q3, and how do they compare to our compliance updates in Q4?", a standard embedding model conflates those distinct concepts. The vector search then returns a messy compromise that misses critical context from both domains. I ran into this exact wall last year while building an internal search engine for cross-departmental documents, and it drove me crazy until we switched to dynamic query decomposition inside an agentic RAG pipeline.

When you deal with complex enterprise search, single-shot retrieval fails. You need an intelligent orchestration layer that breaks a complex prompt into small, focused sub-queries, executes targeted vector search calls for each piece, and synthesizes the retrieved contexts.

Here is how you can implement dynamic query decomposition step by step.

Standard vector search maps an entire user input into a single point in a high-dimensional vector space. When the input contains multiple distinct questions, conditional logic, or temporal comparisons, the resulting vector ends up in a semantic "middle ground." It loses the specific nuances of each sub-topic.

Agentic RAG solves this by introducing dynamic routing and query decomposition. Instead of sending the original user prompt straight to the vector database, an LLM retrieval agent inspects the prompt, identifies distinct information needs, generates targeted sub-queries, and routes them to the appropriate search indexes.

If you work on domain-specific systems such as AI for Accounting & Finance, this approach is mandatory. Financial audits and regulatory compliance questions almost always require pulling data across distinct quarters, departments, and policy documents simultaneously.

Step 1: Define the Structured Output Schemas

First, set up your Python environment. You need Pydantic to enforce strict structured outputs from your LLM decomposer.

from pydantic import BaseModel, Field
from typing import List

class SubQuery(BaseModel):
    id: int = Field(..., description="Unique index for the sub-query")
    query: str = Field(..., description="The specific, standalone query text to search for")
    target_collection: str = Field(..., description="The target vector index or collection, e.g., 'financial_reports', 'legal_docs', or 'ops_logs'")

class QueryDecompositionPlan(BaseModel):
    original_query: str = Field(..., description="The initial query provided by the user")
    sub_queries: List[SubQuery] = Field(..., description="List of decomposed sub-queries needed to answer the original prompt completely")
    reasoning: str = Field(..., description="Brief explanation of why the query was broken down this way")

Using Pydantic ensures your decomposer outputs valid JSON every single time, which prevents parse errors in production.

Step 2: Build the LLM Query Decomposer

Now write the core decomposition engine. This module takes a complex prompt and uses an LLM to split it into clean, isolated vector search queries.

import os
from openai import OpenAI

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

DECOMPOSER_SYSTEM_PROMPT = """
You are an expert query planning agent for an enterprise retrieval system.
Your job is to break down complex, multi-part user prompts into concise, independent sub-queries.

Rules:
1. Each sub-query must be completely self-contained. Replace ambiguous pronouns with explicit nouns.
2. Identify the target document collection for each sub-query ('finance', 'legal', 'operations', or 'general').
3. Keep the number of sub-queries minimal, typically between 2 and 4.
4. If the user query is already simple and single-topic, return a list containing only the original query.
"""

def decompose_query(user_prompt: str) -> QueryDecompositionPlan:
    response = client.beta.chat.completions.parse(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": DECOMPOSER_SYSTEM_PROMPT},
            {"role": "user", "content": user_prompt}
        ],
        response_format=QueryDecompositionPlan,
        temperature=0.0
    )
    return response.choices[0].message.parsed

# Example usage
prompt = "Compare our Q3 legal liabilities with the new Q4 compliance requirements for GDPR."
plan = decompose_query(prompt)

for sq in plan.sub_queries:
    print(f"[{sq.id}] Collection: {sq.target_collection} | Query: {sq.query}")

If you test this with the sample prompt, the model splits the request into two clean searches: one for Q3 legal liabilities in the legal store, and another for Q4 GDPR compliance rules.

Step 3: Run Parallel Vector Search Operations

Once you have your clean sub-queries, execute them concurrently. Running vector searches sequentially adds unnecessary latency to your application.

import asyncio
from typing import Dict, List

# Dummy representation of a vector store lookup
async def execute_vector_search(sub_query: SubQuery) -> Dict:
    # In practice, call your vector DB here (e.g., Qdrant, Pinecone, Weaviate)
    await asyncio.sleep(0.2)  # Simulate network latency

    fake_results = {
        "finance": "Q3 net income increased by 14% quarter-over-quarter.",
        "legal": "Q3 pending litigation costs were estimated at $1.2M.",
        "operations": "GDPR compliance framework updated in Q4 with mandatory 72-hour breach notifications."
    }

    retrieved_text = fake_results.get(sub_query.target_collection, "No matching records found.")

    return {
        "sub_query_id": sub_query.id,
        "query": sub_query.query,
        "collection": sub_query.target_collection,
        "retrieved_context": retrieved_text
    }

async def run_parallel_retrieval(plan: QueryDecompositionPlan) -> List[Dict]:
    tasks = [execute_vector_search(sq) for sq in plan.sub_queries]
    results = await asyncio.gather(*tasks)
    return results

This asynchronous approach scales nicely. Whether your agent generates two sub-queries or four, the total retrieval time remains roughly equal to a single query execution.

If your team is scaling these architectures across enterprise teams, setting up proper pipeline foundations matters early on. Dedicated engineering resources specializing in AI agent development can help optimize these parallel loops to prevent infrastructure bottlenecks.

Step 4: Synthesize Retrieved Contexts into the Final Answer

Now that you have contexts pulled from multiple vector search calls, feed them to a synthesizer LLM. The synthesizer blends the disparate pieces of context into one coherent response.

def synthesize_final_answer(original_query: str, retrieval_results: List[Dict]) -> str:
    formatted_contexts = ""
    for res in retrieval_results:
        formatted_contexts += f"\n--- Context from Query: '{res['query']}' (Index: {res['collection']}) ---\n"
        formatted_contexts += f"{res['retrieved_context']}\n"

    system_prompt = """
    You are an enterprise AI assistant. Answer the user's question using ONLY the provided retrieved contexts.
    If the context items contain distinct pieces of data, compare and synthesize them logically.
    Acknowledge any gaps if the information is incomplete.
    """

    user_content = f"Original Question: {original_query}\n\nRetrieved Contexts:\n{formatted_contexts}"

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_content}
        ],
        temperature=0.2
    )

    return response.choices[0].message.content

# Putting it together
async def main():
    user_prompt = "What were our Q3 legal liabilities and how do they impact our Q4 operational changes?"

    print("Decomposing query...")
    plan = decompose_query(user_prompt)

    print("Executing vector searches...")
    results = await run_parallel_retrieval(plan)

    print("Synthesizing final answer...")
    final_output = synthesize_final_answer(user_prompt, results)

    print("\nFinal Answer:\n", final_output)

# Run loop
asyncio.run(main())

Step 5: Add an Agentic Verification Loop

The real magic of agentic RAG happens when you add reflection. What if the vector search returns garbage for one of the sub-queries? A basic pipeline fails silently. An agentic system evaluates the retrieved context and dynamically retries with a modified search strategy.

Here is a simple evaluation node to handle missing data:

class RetrievalEvaluation(BaseModel):
    is_sufficient: bool = Field(..., description="True if retrieved contexts contain enough info to answer the prompt")
    missing_information: str = Field("", description="Description of missing details if context is insufficient")
    suggested_rewrites: List[str] = Field(default_factory=list, description="Alternative queries to try if search failed")

def evaluate_retrieval(original_query: str, retrieval_results: List[Dict]) -> RetrievalEvaluation:
    eval_prompt = f"""
    Evaluate if the following retrieved contexts contain enough information to answer this prompt:
    Prompt: {original_query}

    Contexts: {retrieval_results}
    """

    response = client.beta.chat.completions.parse(
        model="gpt-4o",
        messages=[{"role": "user", "content": eval_prompt}],
        response_format=RetrievalEvaluation,
        temperature=0.0
    )
    return response.choices[0].message.parsed

If is_sufficient returns False, your orchestrator loop sends suggested_rewrites back to the vector search layer before presenting an answer to the user. This simple feedback loop turns static RAG into a resilient software system.

Wrapping Up

Building enterprise search that actually works requires moving past simple single-prompt embeddings. Dynamic query decomposition breaks complex prompts into manageable tasks, executes vector searches in parallel, and merges contexts cleanly.

Building custom agentic pipelines at scale often requires dedicated engineering expertise to handle production edge cases, latency budgets, and specialized domain integrations. If your organization needs help bringing production-ready AI systems to life, platforms like Gaper connect you with vetted talent to accelerate your technical roadmap.