다단계 조사 루프 설계

리서치 반복 과정이란 무엇입니까?

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단계: 웹 검색

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단계: 사실 읽기 및 추출

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단계: 지식 공백 식별

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

사실 중복 제거

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

전체 리서치 반복 과정

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}

출처 추적

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']})")

속도를 위한 병렬 검색

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

새로운 정보가 발견되지 않을 때 리서치 반복 과정을 종료하는 것은 무엇입니까?

리서치 반복 과정 설계 복습