ディープリサーチループのパターン

ディープリサーチループとは

ループアーキテクチャの概要

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)

フェーズ1：初期検索

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

フェーズ2：重要な事実の抽出

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)

フェーズ3：知識の不足部分の特定

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)

フェーズ4：追加クエリの生成

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

終了条件

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

フェーズ5：最終回答の統合

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

反復をまたいだ情報源の追跡

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)

検索の並列実行

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

事実の重複除去

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

知識チェック

復習：ディープリサーチループパターン