Patrón de ciclo de investigación profunda

¿Qué es el ciclo de investigación profunda?

Descripción general de la arquitectura 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: búsqueda 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: extracción de datos clave

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: identificación de lagunas de conocimiento

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: generación de consultas de seguimiento

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

Criterios de terminación

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íntesis de la respuesta 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
    ))

Seguimiento de fuentes entre iteraciones

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)

Ejecución de búsquedas en 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))

Eliminación de duplicados entre datos

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

Comprobación de conocimientos

Repaso: patrón de ciclo de investigación profunda