Pattern del ciclo di ricerca approfondita

Che cos'è il ciclo di ricerca approfondita?

Panoramica dell'architettura del ciclo

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)

Fase 1: ricerca iniziale

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

Fase 2: estrazione dei fatti chiave

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)

Fase 3: individuazione delle lacune nelle conoscenze

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)

Fase 4: generazione delle query successive

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

Criteri di terminazione

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))

Fase 5: sintesi della risposta 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
    ))

Tracciamento delle fonti tra le iterazioni

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)

Esecuzione parallela delle ricerche

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))

Deduplicazione dei fatti

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

Verifica delle conoscenze

Riepilogo: schema del ciclo di ricerca approfondita