Schéma de boucle de recherche approfondie

Qu'est-ce que la boucle de recherche approfondie ?

Vue d'ensemble de l'architecture de la boucle

def deep_research(question, max_iterations=3):
    all_facts = []
    queries_used = [question]

    for iteration in range(max_iterations):
        results = search_and_rank(queries_used[-1])
        facts = extract_facts(results, question)
        all_facts.extend(facts)

        gaps = identify_knowledge_gaps(question, all_facts)
        if not gaps:
            print(f'Research complete after {iteration + 1} iterations')
            break

        follow_up = generate_follow_up_query(gaps)
        queries_used.append(follow_up)
        print(f'Iteration {iteration + 1}: {follow_up}')

    return synthesize_answer(question, all_facts)

Phase 1 : recherche initiale

from tavily import TavilyClient
import os

client = TavilyClient(api_key=os.getenv('TAVILY_API_KEY'))

def initial_search(question, n_results=5):
    results = client.search(
        query=question,
        max_results=n_results,
        search_depth='advanced'  # deeper extraction for research tasks
    )
    return results.get('results', [])

# Also capture the direct answer if available
def get_direct_answer(question):
    results = client.search(query=question, max_results=3)
    return results.get('answer', '')  # Tavily's synthesized answer

Phase 2 : extraction des faits essentiels

EXTRACT_FACTS_PROMPT = '''You are a research assistant.
From the search results below, extract all factual claims relevant to the question.

Question: {question}

Search results:
{results_text}

Return a JSON list of facts:
["GPT-4 was released in March 2023",
 "GPT-4 supports 128k context window",
 ...]

Only include verifiable facts. No opinions or summaries. JSON:'''

def extract_facts(results, question):
    import json
    results_text = '\n\n'.join(
        f"[{i+1}] {r['title']}\n{r['content'][:600]}"
        for i, r in enumerate(results)
    )
    response = llm_call(EXTRACT_FACTS_PROMPT.format(
        question=question, results_text=results_text
    ))
    return json.loads(response)

Phase 3 : repérage des lacunes dans les connaissances

GAPS_PROMPT = '''You are a research analyst.

Original research question: {question}

Facts gathered so far:
{facts}

What important aspects of the question are still NOT covered by these facts?
List 2-3 specific knowledge gaps as a JSON array of strings.
If the question is fully answered, return an empty array [].

JSON:'''

def identify_knowledge_gaps(question, facts):
    import json
    facts_text = '\n'.join(f'- {f}' for f in facts)
    response = llm_call(GAPS_PROMPT.format(
        question=question, facts=facts_text
    ))
    return json.loads(response)

Phase 4 : génération des requêtes de suivi

QUERY_GEN_PROMPT = '''Convert these knowledge gaps into specific web search queries.

Original topic: {topic}
Knowledge gaps:
{gaps}

Return a JSON array of search query strings.
Make each query specific and searchable.
JSON:'''

def generate_follow_up_queries(topic, gaps):
    import json
    gaps_text = '\n'.join(f'- {g}' for g in gaps)
    response = llm_call(QUERY_GEN_PROMPT.format(
        topic=topic, gaps=gaps_text
    ))
    queries = json.loads(response)
    return queries[:3]  # max 3 follow-up queries per iteration

Critères d'arrêt

MAX_ITERATIONS = 4
MAX_FACTS = 50
MIN_NEW_FACTS_TO_CONTINUE = 3

def should_continue(gaps, all_facts, new_facts, iteration):
    if iteration >= MAX_ITERATIONS:
        print(f'Stopping: max iterations ({MAX_ITERATIONS}) reached')
        return False
    if not gaps:
        print('Stopping: no knowledge gaps found')
        return False
    if len(all_facts) >= MAX_FACTS:
        print(f'Stopping: fact budget ({MAX_FACTS}) reached')
        return False
    if len(new_facts) < MIN_NEW_FACTS_TO_CONTINUE:
        print(f'Stopping: diminishing returns ({len(new_facts)} new facts)')
        return False
    return True

if __name__ == '__main__':
    print('Continue?', should_continue(gaps=['gap1'], all_facts=['f'] * 10, new_facts=['f1', 'f2', 'f3'], iteration=1))
    print('Continue?', should_continue(gaps=[], all_facts=['f'] * 10, new_facts=['f1'], iteration=1))

Phase 5 : synthèse de la réponse finale

SYNTHESIS_PROMPT = '''You are a research writer.

Original question: {question}

Researched facts (gathered from {n_iterations} search iterations):
{facts}

Write a comprehensive, well-structured answer based on these facts.
Format with clear sections. Be specific and cite facts where relevant.
If any aspect of the question could not be fully answered, say so.

Answer:'''

def synthesize_answer(question, all_facts, n_iterations):
    facts_text = '\n'.join(f'- {f}' for f in all_facts)
    return llm_call(SYNTHESIS_PROMPT.format(
        question=question,
        facts=facts_text,
        n_iterations=n_iterations
    ))

Suivi des sources au fil des itérations

def extract_facts_with_sources(results, question):
    import json

    PROMPT = '''Extract facts from these search results. For each fact include the source URL.

Question: {question}
Results: {results_text}

Return JSON list:
[{{"fact": "GPT-4 supports 128k context", "source": "https://openai.com/..."}}, ...]
JSON:'''

    results_text = '\n\n'.join(
        f"[URL: {r['url']}]\n{r['content'][:500]}"
        for r in results
    )
    response = llm_call(PROMPT.format(
        question=question, results_text=results_text
    ))
    return json.loads(response)

Exécution parallèle des recherches

import asyncio

async def async_search(client, query):
    # Tavily has async support
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(
        None,
        lambda: client.search(query=query, max_results=3)
    )
    return result.get('results', [])

async def parallel_search(queries):
    tasks = [async_search(client, q) for q in queries]
    results_list = await asyncio.gather(*tasks)
    # Flatten
    return [r for results in results_list for r in results]

# Run in event loop
all_results = asyncio.run(parallel_search(follow_up_queries))

Déduplication des faits

def deduplicate_facts(facts, similarity_threshold=0.92):
    if not facts:
        return facts

    # Get embeddings for all facts
    embeddings = [embed(f) for f in facts]
    unique_indices = [0]  # always keep first

    for i in range(1, len(facts)):
        is_duplicate = False
        for j in unique_indices:
            sim = cosine_similarity(embeddings[i], embeddings[j])
            if sim > similarity_threshold:
                is_duplicate = True
                break
        if not is_duplicate:
            unique_indices.append(i)

    unique_facts = [facts[i] for i in unique_indices]
    print(f'Deduplicated: {len(facts)} -> {len(unique_facts)} facts')
    return unique_facts

Vérification des connaissances

Récapitulatif : modèle de boucle de recherche approfondie