Deep Research & Agentic RAG

# implement decompose_query
# Key operations: evaluation logic, API interaction

from openai import OpenAI
import json
import asyncio

client = OpenAI()

def decompose_query(query):
 """Break a complex question into sub-queries."""
 response = client.chat.completions.create(
 model="gpt-4o",
 messages=[{
 "role": "system",
 "content": """Decompose the user's research question into
sub-queries. Return JSON with:
- "sub_queries": list of specific, searchable questions
- "dependencies": dict mapping query index to indices
 it depends on (empty list if independent)
- "strategy": "parallel" or "sequential"

Keep sub-queries focused and searchable."""
 }, {
 "role": "user",
 "content": query
 }],
 response_format={"type": "json_object"}
 )
 return json.loads(response.choices[0].message.content)

# Example usage
plan = decompose_query(
 "How do carbon tax policies in the EU and US compare, "
 "and what evidence exists for their effectiveness?"
)
# Returns sub-queries like:
# 1. "What are the current carbon tax policies in the EU?"
# 2. "What are the current carbon tax policies in the US?"
# 3. "What studies evaluate EU carbon tax effectiveness?"
# 4. "What studies evaluate US carbon pricing effectiveness?"

# Implementation example
# See inline comments for step-by-step details.

import asyncio
from typing import List, Dict

async def search_web(query: str) -> List[Dict]:
 """Search the web using a search API."""
 # Implementation with Tavily, Serper, or Brave Search
 pass

async def search_documents(query: str, collection) -> List[Dict]:
 """Search internal document store."""
 results = collection.query(query_texts=[query], n_results=5)
 return [{"text": d, "source": "internal_docs"}
 for d in results["documents"][0]]

async def search_database(query: str) -> List[Dict]:
 """Convert query to SQL and search database."""
 # Text-to-SQL pipeline (covered in Section 20.5)
 pass

async def multi_source_search(sub_queries: List[str], collection):
 """Execute sub-queries in parallel across sources."""
 all_results = {}

 async def search_one(query):
 # Search all sources in parallel for each query
 web, docs, db = await asyncio.gather(
 search_web(query),
 search_documents(query, collection),
 search_database(query),
 return_exceptions=True
 )
 return {
 "query": query,
 "web": web if not isinstance(web, Exception) else [],
 "docs": docs if not isinstance(docs, Exception) else [],
 "db": db if not isinstance(db, Exception) else []
 }

 # Execute all sub-queries in parallel
 tasks = [search_one(q) for q in sub_queries]
 results = await asyncio.gather(*tasks)
 return results

# implement evaluate_and_refine
# Key operations: retrieval pipeline, evaluation logic, API interaction

def evaluate_and_refine(original_query, gathered_info, max_iterations=3):
 """Iteratively refine retrieval until sufficient."""

 for iteration in range(max_iterations):
 # Ask the LLM to evaluate sufficiency
 eval_response = client.chat.completions.create(
 model="gpt-4o",
 messages=[{
 "role": "system",
 "content": """Evaluate whether the gathered information is
sufficient to comprehensively answer the question.
Return JSON with:
- "sufficient": true/false
- "missing": list of what information is still needed
- "follow_up_queries": list of queries to fill gaps
- "confidence": 0.0 to 1.0"""
 }, {
 "role": "user",
 "content": f"""Original question: {original_query}

Gathered information:
{json.dumps(gathered_info, indent=2)}"""
 }],
 response_format={"type": "json_object"}
 )

 evaluation = json.loads(
 eval_response.choices[0].message.content
 )

 if evaluation["sufficient"] or evaluation["confidence"] > 0.85:
 return gathered_info, evaluation

 # Execute follow-up queries
 follow_ups = evaluation["follow_up_queries"]
 new_info = retrieve_for_queries(follow_ups)
 gathered_info.extend(new_info)

 return gathered_info, evaluation

# implement synthesize_research
# Key operations: API interaction

def synthesize_research(original_query, gathered_info, credibility_scores):
 """Synthesize gathered information into a research report."""

 # Sort by credibility, place highest-trust sources first
 sorted_info = sorted(
 zip(gathered_info, credibility_scores),
 key=lambda x: x[1],
 reverse=True
 )

 context = "\n\n".join([
 f"[Source {i+1} | Credibility: {score:.1f}/5]\n{info['text']}"
 for i, (info, score) in enumerate(sorted_info)
 ])

 response = client.chat.completions.create(
 model="gpt-4o",
 messages=[{
 "role": "system",
 "content": """You are a research analyst. Synthesize the
provided sources into a comprehensive answer.

Guidelines:
- Cite sources by number [Source N]
- Note where sources agree or disagree
- Flag claims supported by only one source
- Prioritize high-credibility sources
- Acknowledge gaps in the evidence
- Structure the answer with clear sections"""
 }, {
 "role": "user",
 "content": f"""Research Question: {original_query}

Sources:
{context}"""
 }],
 temperature=0.2
 )

 return response.choices[0].message.content