Padrão de ciclo de pesquisa aprofundada

O que é o ciclo de pesquisa aprofundada

Visão geral da arquitetura do 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: pesquisa inicial

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: extração dos fatos essenciais

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: identificação das lacunas de conhecimento

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: geração de consultas de acompanhamento

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érios de encerramento

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: síntese da resposta final

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

Rastreamento de fontes entre iterações

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)

Execução de pesquisas em paralelo

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

Deduplicação de fatos

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ção de conhecimento

Recapitulação: padrão de ciclo de pesquisa aprofundada