0Pricing
AI Agents · レッスン

複数ステップのリサーチループ設計

計画 → 検索 → 読み取り → 抽出 → 統合を、十分な深さに達するまで繰り返します。

「複数ステップのリサーチループ設計」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン1/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。

調査ループとは

調査ループとは、LLMが計画、検索、読み取り、抽出、ギャップの特定、反復を行い、調査の目標が達成されるまで続けるエージェント型のパターンです。

単発の検索とは異なり、このループは見つかった内容に応じて適応します。予想外の手がかりを追い、行き止まりになった手がかりは破棄します。

フェーズ1:質問の分解

最初のステップは、調査の質問をサブ質問に分解することです。これにより、方向性のある検索計画を作成でき、エージェントが目的なく閲覧するのを防げます。

import openai, json

client = openai.OpenAI(api_key='YOUR_OPENAI_KEY')

def decompose_question(question: str) -> list[str]:
    prompt = (
        f'Break this research question into 3-5 focused sub-questions.\n'
        f'Each sub-question should be independently searchable.\n'
        f'Question: "{question}"\n'
        f'Return JSON: {{"sub_questions": ["..."]}}'
    )
    resp = client.chat.completions.create(
        model='gpt-4o',
        messages=[{'role': 'user', 'content': prompt}],
        response_format={'type': 'json_object'}
    )
    return json.loads(resp.choices[0].message.content)['sub_questions']

questions = decompose_question('What are the main causes of inflation in 2024?')
print(questions)

フェーズ2:ウェブ検索

各サブ質問について、ウェブ検索を実行します。検索API(Serper、Brave、Bing)を使って、URLとスニペットのリストを取得します。すべてのページを読むのではなく、まず高品質な情報源を優先します。

import requests

SERPER_KEY = 'YOUR_SERPER_API_KEY'

def search_web(query: str, num_results: int = 5) -> list[dict]:
    resp = requests.post(
        'https://google.serper.dev/search',
        headers={'X-API-KEY': SERPER_KEY, 'Content-Type': 'application/json'},
        json={'q': query, 'num': num_results}
    )
    resp.raise_for_status()
    results = resp.json().get('organic', [])
    return [
        {'title': r['title'], 'url': r['link'], 'snippet': r.get('snippet', '')}
        for r in results
    ]

フェーズ3:読み取りと事実の抽出

各URLを取得し、サブ質問に関連する重要な事実を抽出します。LLMを使って記事を読み取り、情報源のURL付きで事実のリストを作成します。

import httpx
from bs4 import BeautifulSoup

def fetch_text(url: str, max_chars: int = 4000) -> str:
    try:
        resp = httpx.get(url, timeout=10, follow_redirects=True,
                         headers={'User-Agent': 'ResearchAgent/1.0'})
        soup = BeautifulSoup(resp.text, 'html.parser')
        for tag in soup(['script', 'style', 'nav', 'footer']):
            tag.decompose()
        return soup.get_text(separator=' ', strip=True)[:max_chars]
    except Exception:
        return ''

def extract_facts(text: str, question: str, url: str) -> list[dict]:
    prompt = (
        f'Extract key facts from the text that answer: "{question}"\n'
        f'Return JSON: {{"facts": ["fact1", ...]}}\n\n'
        f'TEXT:\n{text[:3000]}'
    )
    resp = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        response_format={'type': 'json_object'}
    )
    facts = json.loads(resp.choices[0].message.content).get('facts', [])
    return [{'fact': f, 'source': url} for f in facts]

フェーズ4:知識のギャップの特定

読み取りが終わったら、LLMに蓄積した事実を確認させ、まだ不明な点を特定します。これらのギャップが、次の検索クエリになります。

def identify_gaps(original_question: str, facts: list[dict]) -> list[str]:
    fact_text = '\n'.join(f'- {f["fact"]}' for f in facts[:20])
    prompt = (
        f'Original question: "{original_question}"\n'
        f'Facts gathered so far:\n{fact_text}\n\n'
        f'What key aspects are still unanswered? '
        f'Return 0-3 follow-up search queries (0 if research is complete).\n'
        f'JSON: {{"gaps": ["search query 1", ...]}}'
    )
    resp = client.chat.completions.create(
        model='gpt-4o',
        messages=[{'role': 'user', 'content': prompt}],
        response_format={'type': 'json_object'}
    )
    return json.loads(resp.choices[0].message.content).get('gaps', [])

終了条件

ループを無限に実行しないよう、明確な停止条件が必要です。次の場合に停止します。

  • 新たなギャップが特定されない
  • 直前の反復で新しい事実が追加されなかった
  • 最大反復回数に達した(安全上の上限)
  • 事実の総数がしきい値を超えた(十分な深さに達した)
def should_continue(gaps: list[str], new_facts_this_round: int,
                    iteration: int, total_facts: int) -> bool:
    if iteration >= 5:
        return False          # Hard cap: 5 iterations
    if new_facts_this_round == 0:
        return False          # No new information found
    if not gaps:
        return False          # LLM says research is complete
    if total_facts >= 50:
        return False          # Sufficient depth reached
    return True

if __name__ == '__main__':
    print('Continue (has gaps, new facts)?', should_continue(['gap1'], 4, iteration=1, total_facts=10))
    print('Continue (no new facts)?', should_continue(['gap1'], 0, iteration=1, total_facts=10))

事実の重複排除

複数の情報源が同じ事実を報告していることはよくあります。LLMに意味的に同等の事実を統合させ、最も信頼できる情報源に基づくものを残して重複を排除します。

def deduplicate_facts(facts: list[dict]) -> list[dict]:
    if len(facts) <= 3:
        return facts
    fact_text = '\n'.join(
        f'{i}: {f["fact"]} (source: {f["source"]})' for i, f in enumerate(facts)
    )
    prompt = (
        f'Remove duplicate or near-duplicate facts. Keep the most informative version.\n'
        f'Return JSON: {{"keep_indices": [0, 1, ...]}}\n\n'
        f'FACTS:\n{fact_text}'
    )
    resp = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        response_format={'type': 'json_object'}
    )
    indices = json.loads(resp.choices[0].message.content).get('keep_indices', [])
    return [facts[i] for i in indices if i < len(facts)]

調査ループ全体

すべてのフェーズを1つの research() 関数にまとめ、終了条件に達するまでループを実行します。

def research(question: str) -> dict:
    all_facts = []
    iteration = 0

    # Phase 1: Decompose into sub-questions
    queries = decompose_question(question)

    while True:
        new_facts_this_round = 0

        for query in queries:
            results = search_web(query, num_results=3)
            for result in results:
                text = fetch_text(result['url'])
                if not text:
                    continue
                facts = extract_facts(text, query, result['url'])
                all_facts.extend(facts)
                new_facts_this_round += len(facts)

        all_facts = deduplicate_facts(all_facts)
        gaps = identify_gaps(question, all_facts)
        iteration += 1

        if not should_continue(gaps, new_facts_this_round, iteration, len(all_facts)):
            break

        queries = gaps  # next iteration searches for the gaps

    return {'facts': all_facts, 'iterations': iteration}

出典の追跡

最終レポートに引用を含められるよう、すべての事実に出典URLを付ける必要があります。処理中に出典メタデータを削除してはいけません。引用レイヤーで必要になるためです。

def add_fact(fact_list: list, fact_text: str, source_url: str, iteration: int):
    fact_list.append({
        'fact':      fact_text,
        'source':    source_url,
        'iteration': iteration,
        'verified':  False   # set to True after cross-referencing
    })

# Example:
facts_store = []
add_fact(facts_store, 'Global inflation peaked at 9.1% in June 2022',
         'https://bls.gov/news.release/cpi.htm', iteration=1)

print(f'Logged {len(facts_store)} fact(s):')
for f in facts_store:
    print(f"  - {f['fact']} (source: {f['source']})")

高速化のための並列検索

検索を順番に実行すると時間がかかります。5個のクエリ × 3個のURL × 1回の取得で、HTTP呼び出しが15回順番に発生します。concurrent.futuresを使って、各反復内の取得を並列化します。

from concurrent.futures import ThreadPoolExecutor, as_completed

def parallel_research_round(queries: list[str]) -> list[dict]:
    collected = []

    def process_query(query):
        results = search_web(query, num_results=3)
        facts = []
        for r in results:
            text = fetch_text(r['url'])
            if text:
                facts.extend(extract_facts(text, query, r['url']))
        return facts

    with ThreadPoolExecutor(max_workers=4) as ex:
        futures = {ex.submit(process_query, q): q for q in queries}
        for future in as_completed(futures):
            collected.extend(future.result())

    return collected

ループの進行状況の監視

ループが停止した理由や、想定以上に長く実行された理由をデバッグできるよう、各反復をログに記録します。クエリ数、追加された事実、特定されたギャップを含めます。

import logging

log = logging.getLogger('research_loop')
import sys
logging.basicConfig(level=logging.INFO, stream=sys.stdout)

def log_iteration(iteration: int, queries: list[str],
                  new_facts: int, total_facts: int, gaps: list[str]):
    log.info(
        'Iteration %d | Queries: %d | New facts: %d | Total: %d | Gaps: %d',
        iteration, len(queries), new_facts, total_facts, len(gaps)
    )
    if gaps:
        for g in gaps:
            log.debug('  Gap query: %s', g)

if __name__ == '__main__':
    log_iteration(iteration=2, queries=['who are our top competitors?'], new_facts=4, total_facts=12, gaps=['pricing data'])

新しい情報が見つからないとき、調査ループを終了させるものは何ですか

終了条件を理解すると、無限ループを防ぎ、妥当な時間内にエージェントが結果を返せるようになります。

調査ループ設計の振り返り

複数ステップの調査ループは、分解 → 検索 → 読み取り → 抽出 → ギャップの特定 → 反復という流れで進みます。ギャップがなくなったとき、新しい事実が見つからなくなったとき、または反復回数の上限に達したときに停止します。

すべての事実に出典URLを付け、検索を並列化して高速化し、統合する前に事実の重複を排除します。

よくある質問

「複数ステップのリサーチループ設計」レッスンは無料ですか?

はい。「複数ステップのリサーチループ設計」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し(組み込みコードエディタと24時間対応のAIチューター)、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。

「複数ステップのリサーチループ設計」で何を学びますか?

計画 → 検索 → 読み取り → 抽出 → 統合を、十分な深さに達するまで繰り返します。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

AI Agentsを始めるのに経験は必要ですか?

事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン1/4です。

「複数ステップのリサーチループ設計」レッスンにはどのくらい時間がかかりますか?

ほとんどのCoddyKitレッスンは約5~10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

このAI Agentsレッスンでコードを書いて実行できますか?

はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

  1. 複数ステップのリサーチループ設計
  2. 情報源の検証と引用
  3. 構造化レポートの生成
  4. ファクトチェックとハルシネーション防止
← AI Agentsに戻る