"Deep Research" Capability - 완벽 핸즈온 가이드

🎯 학습 목표

• Chain-of-Thought (CoT) 프롬프팅 마스터
• Tree of Thoughts (ToT) 구현
• Self-Consistency 기법 적용
• Sequential Revision (순차적 개선)
• 복잡한 다단계 추론 시스템 구축
• OpenAI o1 스타일 추론 시뮬레이션

📋 사전 준비

1. 개발 환경 설정

# 새 프로젝트 디렉토리
mkdir deep-research-agent
cd deep-research-agent

# 가상환경 생성
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 필수 패키지 설치
pip install openai anthropic streamlit python-dotenv
pip install langchain langchain-openai
pip install networkx matplotlib graphviz  # 추론 트리 시각화
pip install tiktoken numpy
pip install tavily-python duckduckgo-search

2. 프로젝트 구조

mkdir -p src reasoning utils visualizations data
touch .env app.py requirements.txt

deep-research-agent/
├── src/
│   ├── cot_reasoner.py          # Chain-of-Thought
│   ├── tot_reasoner.py          # Tree of Thoughts
│   ├── self_consistency.py      # Self-Consistency
│   ├── sequential_revision.py   # 순차적 개선
│   └── deep_research.py         # 통합 시스템
├── reasoning/
│   ├── prompt_templates.py      # 추론 프롬프트
│   ├── verification.py          # 검증 로직
│   └── scoring.py               # 추론 평가
├── utils/
│   ├── search_tools.py          # 검색 도구
│   └── response_parser.py       # 응답 파싱
├── visualizations/
│   └── tree_visualizer.py       # 추론 트리 시각화
├── app.py                       # Streamlit UI
└── .env

---

🧠 Step 1: Chain-of-Thought (CoT) 추론 (25분)

reasoning/prompt_templates.py

class ReasoningPrompts:
    """추론 프롬프트 템플릿"""
    
    ZERO_SHOT_COT = """질문: {question}

단계별로 생각해봅시다."""

    FEW_SHOT_COT = """다음 예시를 참고하여 단계별로 생각하면서 문제를 해결하세요.

예시 1:
질문: 카페에 처음에 23명이 있었습니다. 5명이 나가고 8명이 들어왔습니다. 지금 몇 명이 있나요?
생각: 단계별로 해결해봅시다.
1. 처음에 23명이 있었습니다
2. 5명이 나갔으므로: 23 - 5 = 18명
3. 8명이 들어왔으므로: 18 + 8 = 26명
따라서 답은 26명입니다.

예시 2:
질문: 사과 3개의 가격이 1,500원입니다. 사과 7개의 가격은 얼마인가요?
생각: 단계별로 해결해봅시다.
1. 사과 1개의 가격을 먼저 구합니다: 1,500 ÷ 3 = 500원
2. 사과 7개의 가격: 500 × 7 = 3,500원
따라서 답은 3,500원입니다.

질문: {question}
생각: 단계별로 해결해봅시다."""

    SELF_ASK = """질문: {question}

이 질문에 답하기 위해 필요한 하위 질문들을 먼저 정리하고, 각각에 답한 후 최종 답변을 도출하세요.

하위 질문 1:
답변 1:

하위 질문 2:
답변 2:

...

최종 답변:"""

    TREE_OF_THOUGHTS = """질문: {question}

여러 가능한 해결 방법을 탐색하겠습니다.

접근법 A:
단계 1: 
평가: [1-10점]

접근법 B:
단계 1:
평가: [1-10점]

접근법 C:
단계 1:
평가: [1-10점]

가장 유망한 접근법을 선택하여 계속 진행하세요."""

    VERIFICATION = """다음 답변이 올바른지 검증해주세요.

질문: {question}
답변: {answer}

검증 과정:
1. 논리적 일관성 확인:
2. 사실 확인:
3. 계산 검증:

최종 판단: [올바름/오류있음]
이유:"""

src/cot_reasoner.py

import os
from typing import Dict, List, Optional
from openai import OpenAI
from anthropic import Anthropic
from dotenv import load_dotenv
import sys
sys.path.append('..')

from reasoning.prompt_templates import ReasoningPrompts

load_dotenv()

class ChainOfThoughtReasoner:
    """Chain-of-Thought 추론 엔진"""
    
    def __init__(self, model_type: str = "openai", model: str = "gpt-4o-mini"):
        """
        Args:
            model_type: 'openai' 또는 'anthropic'
            model: 모델 이름
        """
        self.model_type = model_type
        self.model = model
        
        if model_type == "openai":
            self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        else:
            self.client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
    
    def zero_shot_cot(self, question: str) -> Dict:
        """Zero-shot Chain-of-Thought"""
        prompt = ReasoningPrompts.ZERO_SHOT_COT.format(question=question)
        
        if self.model_type == "openai":
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "당신은 논리적 추론에 뛰어난 AI 어시스턴트입니다."},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.7,
                max_tokens=1000
            )
            reasoning = response.choices[0].message.content
        else:
            response = self.client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1000,
                temperature=0.7,
                messages=[{"role": "user", "content": prompt}]
            )
            reasoning = response.content[0].text
        
        return {
            'method': 'zero_shot_cot',
            'question': question,
            'reasoning': reasoning,
            'prompt': prompt
        }
    
    def few_shot_cot(self, question: str) -> Dict:
        """Few-shot Chain-of-Thought (예시 제공)"""
        prompt = ReasoningPrompts.FEW_SHOT_COT.format(question=question)
        
        if self.model_type == "openai":
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "당신은 단계별 추론에 뛰어난 AI입니다."},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.3,
                max_tokens=1000
            )
            reasoning = response.choices[0].message.content
        else:
            response = self.client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1000,
                temperature=0.3,
                messages=[{"role": "user", "content": prompt}]
            )
            reasoning = response.content[0].text
        
        return {
            'method': 'few_shot_cot',
            'question': question,
            'reasoning': reasoning,
            'prompt': prompt
        }
    
    def self_ask(self, question: str) -> Dict:
        """Self-Ask: 하위 질문 생성 및 답변"""
        prompt = ReasoningPrompts.SELF_ASK.format(question=question)
        
        if self.model_type == "openai":
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "하위 질문을 통해 복잡한 문제를 해결하는 AI입니다."},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.5,
                max_tokens=1500
            )
            reasoning = response.choices[0].message.content
        else:
            response = self.client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1500,
                temperature=0.5,
                messages=[{"role": "user", "content": prompt}]
            )
            reasoning = response.content[0].text
        
        return {
            'method': 'self_ask',
            'question': question,
            'reasoning': reasoning,
            'prompt': prompt
        }
    
    def compare_methods(self, question: str) -> Dict:
        """여러 CoT 방법 비교"""
        results = {
            'question': question,
            'zero_shot': self.zero_shot_cot(question),
            'few_shot': self.few_shot_cot(question),
            'self_ask': self.self_ask(question)
        }
        
        return results

# 테스트
if __name__ == "__main__":
    reasoner = ChainOfThoughtReasoner(model_type="openai")
    
    # 복잡한 추론 질문들
    questions = [
        "철수는 사과 5개를 가지고 있었고, 영희에게 2개를 주었습니다. 그런 다음 민수가 철수에게 사과 3개를 주었습니다. 철수는 지금 몇 개의 사과를 가지고 있나요?",
        "2024년에 태어난 아이가 2050년에 몇 살이 될까요?",
        "한 회사의 직원이 100명인데, 그 중 60%가 남성입니다. 남성 직원 중 30%가 관리자라면, 남성 관리자는 몇 명인가요?"
    ]
    
    for question in questions:
        print(f"\n{'='*70}")
        print(f"질문: {question}")
        print(f"{'='*70}\n")
        
        # Zero-shot CoT
        result = reasoner.zero_shot_cot(question)
        print("📝 Zero-shot CoT:")
        print(result['reasoning'])
        print()
        
        # Few-shot CoT
        result = reasoner.few_shot_cot(question)
        print("📚 Few-shot CoT:")
        print(result['reasoning'])
        print()

---

🌳 Step 2: Tree of Thoughts (ToT) 구현 (35분)

src/tot_reasoner.py

import os
from typing import Dict, List, Tuple
from openai import OpenAI
from dotenv import load_dotenv
import json

load_dotenv()

class TreeNode:
    """추론 트리 노드"""
    
    def __init__(self, state: str, parent=None, depth: int = 0):
        self.state = state  # 현재 추론 상태
        self.parent = parent
        self.children = []
        self.depth = depth
        self.value_score = 0.0  # 이 노드의 가치 점수
        self.visit_count = 0
    
    def add_child(self, child_state: str):
        """자식 노드 추가"""
        child = TreeNode(child_state, parent=self, depth=self.depth + 1)
        self.children.append(child)
        return child
    
    def get_path(self) -> List[str]:
        """루트에서 현재 노드까지의 경로"""
        path = []
        node = self
        while node:
            path.append(node.state)
            node = node.parent
        return list(reversed(path))

class TreeOfThoughtsReasoner:
    """Tree of Thoughts 추론 엔진"""
    
    def __init__(self, model: str = "gpt-4o-mini", max_depth: int = 3, branches: int = 3):
        """
        Args:
            model: OpenAI 모델
            max_depth: 최대 트리 깊이
            branches: 각 노드에서 생성할 가지 수
        """
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.model = model
        self.max_depth = max_depth
        self.branches = branches
    
    def generate_thoughts(self, question: str, current_state: str, num_thoughts: int = 3) -> List[str]:
        """현재 상태에서 가능한 다음 생각들 생성"""
        prompt = f"""질문: {question}

현재까지의 추론:
{current_state}

이어서 가능한 {num_thoughts}가지 다음 추론 단계를 생성하세요. 각각은 서로 다른 접근 방식이어야 합니다.

형식:
1. [첫 번째 추론]
2. [두 번째 추론]
3. [세 번째 추론]"""

        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.8,  # 다양성을 위해 높게 설정
            max_tokens=800
        )
        
        text = response.choices[0].message.content
        
        # 번호 매긴 리스트 파싱
        thoughts = []
        for line in text.strip().split('\n'):
            line = line.strip()
            if line and (line[0].isdigit() or line.startswith('-')):
                thought = line.split('.', 1)[-1].strip()
                if thought:
                    thoughts.append(thought)
        
        return thoughts[:num_thoughts]
    
    def evaluate_thought(self, question: str, thought_path: List[str]) -> float:
        """추론 경로의 품질 평가 (0-1 점수)"""
        path_text = "\n".join([f"단계 {i+1}: {t}" for i, t in enumerate(thought_path)])
        
        prompt = f"""질문: {question}

추론 경로:
{path_text}

이 추론 경로가 올바른 답에 얼마나 가까운지 0-10점으로 평가하세요.

평가 기준:
- 논리적 일관성
- 질문과의 관련성
- 진전도
- 완성도

점수만 숫자로 답하세요 (0-10):"""

        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.0,
            max_tokens=10
        )
        
        try:
            score = float(response.choices[0].message.content.strip())
            return min(max(score / 10.0, 0.0), 1.0)  # 0-1로 정규화
        except:
            return 0.5  # 파싱 실패 시 중간값
    
    def bfs_search(self, question: str) -> Dict:
        """BFS (너비 우선 탐색)로 최적 경로 찾기"""
        root = TreeNode("시작: 문제를 분석합니다.")
        queue = [root]
        all_nodes = [root]
        best_node = root
        best_score = 0.0
        
        print(f"🌳 Tree of Thoughts 탐색 시작 (최대 깊이: {self.max_depth})")
        
        while queue and len(all_nodes) < 50:  # 최대 50개 노드
            current = queue.pop(0)
            
            if current.depth >= self.max_depth:
                continue
            
            print(f"\n{'  ' * current.depth}📍 깊이 {current.depth}: {current.state[:50]}...")
            
            # 현재 노드에서 가능한 다음 생각들 생성
            path = current.get_path()
            next_thoughts = self.generate_thoughts(question, "\n".join(path), self.branches)
            
            for thought in next_thoughts:
                child = current.add_child(thought)
                all_nodes.append(child)
                
                # 평가
                child_path = child.get_path()
                score = self.evaluate_thought(question, child_path)
                child.value_score = score
                
                print(f"{'  ' * (current.depth + 1)}├─ 점수: {score:.2f} - {thought[:40]}...")
                
                # 최고 점수 업데이트
                if score > best_score:
                    best_score = score
                    best_node = child
                
                # 큐에 추가
                if current.depth < self.max_depth - 1:
                    queue.append(child)
        
        # 최종 답변 생성
        best_path = best_node.get_path()
        final_answer = self.generate_final_answer(question, best_path)
        
        return {
            'method': 'tree_of_thoughts_bfs',
            'question': question,
            'best_path': best_path,
            'best_score': best_score,
            'total_nodes': len(all_nodes),
            'final_answer': final_answer,
            'tree_nodes': all_nodes
        }
    
    def dfs_search(self, question: str, node: TreeNode = None, depth: int = 0) -> TreeNode:
        """DFS (깊이 우선 탐색)"""
        if node is None:
            node = TreeNode("시작: 문제를 분석합니다.")
        
        if depth >= self.max_depth:
            path = node.get_path()
            node.value_score = self.evaluate_thought(question, path)
            return node
        
        # 다음 생각들 생성
        path = node.get_path()
        next_thoughts = self.generate_thoughts(question, "\n".join(path), self.branches)
        
        best_child = None
        best_score = 0.0
        
        for thought in next_thoughts:
            child = node.add_child(thought)
            result = self.dfs_search(question, child, depth + 1)
            
            if result.value_score > best_score:
                best_score = result.value_score
                best_child = result
        
        return best_child if best_child else node
    
    def generate_final_answer(self, question: str, reasoning_path: List[str]) -> str:
        """추론 경로를 바탕으로 최종 답변 생성"""
        path_text = "\n".join([f"{i+1}. {step}" for i, step in enumerate(reasoning_path)])
        
        prompt = f"""질문: {question}

최적의 추론 경로:
{path_text}

위 추론 과정을 바탕으로 최종 답변을 명확하고 간결하게 작성하세요."""

        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3,
            max_tokens=500
        )
        
        return response.choices[0].message.content

# 테스트
if __name__ == "__main__":
    tot = TreeOfThoughtsReasoner(max_depth=3, branches=3)
    
    questions = [
        "24와 36의 최대공약수를 구하는 과정을 설명하세요.",
        "한 상자에 빨간 공 5개와 파란 공 3개가 있습니다. 무작위로 2개를 뽑을 때, 두 개 모두 빨간 공일 확률은?",
    ]
    
    for question in questions:
        print(f"\n{'#'*70}")
        print(f"질문: {question}")
        print(f"{'#'*70}")
        
        result = tot.bfs_search(question)
        
        print(f"\n{'='*70}")
        print(f"최적 추론 경로 (점수: {result['best_score']:.2f}):")
        print(f"{'='*70}")
        for i, step in enumerate(result['best_path'], 1):
            print(f"{i}. {step}")
        
        print(f"\n{'='*70}")
        print(f"최종 답변:")
        print(f"{'='*70}")
        print(result['final_answer'])
        
        print(f"\n탐색한 총 노드 수: {result['total_nodes']}")

---

🔄 Step 3: Self-Consistency (자기 일관성) (20분)

src/self_consistency.py

import os
from typing import Dict, List
from openai import OpenAI
from collections import Counter
import re
from dotenv import load_dotenv

load_dotenv()

class SelfConsistencyReasoner:
    """Self-Consistency: 여러 추론 경로로 답변 검증"""
    
    def __init__(self, model: str = "gpt-4o-mini", num_samples: int = 5):
        """
        Args:
            model: OpenAI 모델
            num_samples: 생성할 답변 샘플 수
        """
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.model = model
        self.num_samples = num_samples
    
    def generate_reasoning_paths(self, question: str) -> List[Dict]:
        """여러 개의 독립적인 추론 경로 생성"""
        paths = []
        
        prompt = f"""질문: {question}

단계별로 신중하게 생각하여 답하세요."""

        for i in range(self.num_samples):
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[{"role": "user", "content": prompt}],
                temperature=0.7,  # 다양성 확보
                max_tokens=800
            )
            
            reasoning = response.choices[0].message.content
            answer = self.extract_final_answer(reasoning)
            
            paths.append({
                'path_id': i + 1,
                'reasoning': reasoning,
                'answer': answer
            })
            
            print(f"경로 {i+1} 생성 완료: {answer}")
        
        return paths
    
    def extract_final_answer(self, text: str) -> str:
        """추론에서 최종 답변 추출"""
        # "답은", "답:", "최종 답변" 등을 찾음
        patterns = [
            r'답은?\s*:?\s*(.+?)(?:\.|$)',
            r'최종\s*답변\s*:?\s*(.+?)(?:\.|$)',
            r'따라서\s*(.+?)(?:\.|$)',
            r'그러므로\s*(.+?)(?:\.|$)',
        ]
        
        for pattern in patterns:
            match = re.search(pattern, text, re.IGNORECASE)
            if match:
                return match.group(1).strip()
        
        # 패턴 매칭 실패 시 마지막 문장 반환
        sentences = text.strip().split('.')
        return sentences[-1].strip() if sentences else text
    
    def find_consensus(self, paths: List[Dict]) -> Dict:
        """다수결로 최종 답변 결정"""
        answers = [path['answer'] for path in paths]
        
        # 답변 정규화 (공백, 대소문자 무시)
        normalized_answers = [ans.lower().strip() for ans in answers]
        
        # 가장 많이 나온 답변 찾기
        answer_counts = Counter(normalized_answers)
        most_common = answer_counts.most_common(1)[0]
        consensus_answer = most_common[0]
        consensus_count = most_common[1]
        confidence = consensus_count / len(paths)
        
        return {
            'consensus_answer': consensus_answer,
            'confidence': confidence,
            'vote_distribution': dict(answer_counts),
            'all_answers': answers
        }
    
    def reason_with_consistency(self, question: str) -> Dict:
        """Self-Consistency 추론 실행"""
        print(f"🔄 {self.num_samples}개의 추론 경로 생성 중...\n")
        
        # 1. 여러 추론 경로 생성
        paths = self.generate_reasoning_paths(question)
        
        # 2. 합의 답변 찾기
        print(f"\n📊 답변 집계 중...")
        consensus = self.find_consensus(paths)
        
        # 3. 최종 설명 생성
        final_explanation = self.generate_final_explanation(
            question,
            paths,
            consensus['consensus_answer']
        )
        
        return {
            'method': 'self_consistency',
            'question': question,
            'num_paths': len(paths),
            'paths': paths,
            'consensus': consensus,
            'final_answer': consensus['consensus_answer'],
            'confidence': consensus['confidence'],
            'explanation': final_explanation
        }
    
    def generate_final_explanation(self, question: str, paths: List[Dict], answer: str) -> str:
        """최종 설명 생성"""
        reasoning_summary = "\n\n".join([
            f"경로 {p['path_id']}:\n{p['reasoning'][:200]}..."
            for p in paths[:3]  # 처음 3개만
        ])
        
        prompt = f"""질문: {question}

여러 추론 경로의 요약:
{reasoning_summary}

최종 합의 답변: {answer}

위 추론 경로들을 종합하여 최종 답변에 대한 명확한 설명을 작성하세요."""

        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3,
            max_tokens=500
        )
        
        return response.choices[0].message.content

# 테스트
if __name__ == "__main__":
    reasoner = SelfConsistencyReasoner(num_samples=5)
    
    questions = [
        "12와 18의 최소공배수는 무엇인가요?",
        "100미터를 10초에 달리는 사람의 속도는 시속 몇 km인가요?",
    ]
    
    for question in questions:
        print(f"\n{'#'*70}")
        print(f"질문: {question}")
        print(f"{'#'*70}\n")
        
        result = reasoner.reason_with_consistency(question)
        
        print(f"\n{'='*70}")
        print(f"답변 분포:")
        print(f"{'='*70}")
        for ans, count in result['consensus']['vote_distribution'].items():
            print(f"  '{ans}': {count}표")
        
        print(f"\n{'='*70}")
        print(f"최종 답변 (신뢰도: {result['confidence']:.1%}):")
        print(f"{'='*70}")
        print(result['final_answer'])
        
        print(f"\n{'='*70}")
        print(f"설명:")
        print(f"{'='*70}")
        print(result['explanation'])

---

🔧 Step 4: Sequential Revision (순차적 개선) (20분)

src/sequential_revision.py

import os
from typing import Dict, List
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

class SequentialRevisionReasoner:
    """Sequential Revision: 반복적으로 답변 개선"""
    
    def __init__(self, model: str = "gpt-4o-mini", max_revisions: int = 3):
        """
        Args:
            model: OpenAI 모델
            max_revisions: 최대 수정 횟수
        """
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.model = model
        self.max_revisions = max_revisions
    
    def generate_initial_answer(self, question: str) -> str:
        """초기 답변 생성"""
        prompt = f"""질문: {question}

이 질문에 답변해주세요."""

        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,
            max_tokens=800
        )
        
        return response.choices[0].message.content
    
    def critique_answer(self, question: str, answer: str) -> Dict:
        """답변 비평 및 개선점 도출"""
        prompt = f"""질문: {question}

현재 답변:
{answer}

이 답변을 비평하고 개선할 점을 지적해주세요.

비평 형식:
강점:
- [강점 1]
- [강점 2]

약점:
- [약점 1]
- [약점 2]

개선 제안:
- [제안 1]
- [제안 2]"""

        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.5,
            max_tokens=600
        )
        
        critique = response.choices[0].message.content
        
        return {
            'critique': critique,
            'has_issues': '약점' in critique or '문제' in critique
        }
    
    def revise_answer(self, question: str, current_answer: str, critique: str) -> str:
        """비평을 바탕으로 답변 개선"""
        prompt = f"""질문: {question}

현재 답변:
{current_answer}

비평:
{critique}

위 비평을 반영하여 답변을 개선해주세요. 더 정확하고, 완전하고, 명확한 답변을 작성하세요."""

        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.5,
            max_tokens=1000
        )
        
        return response.choices[0].message.content
    
    def revise_iteratively(self, question: str) -> Dict:
        """반복적 개선 프로세스"""
        revisions = []
        
        # 1. 초기 답변
        print("📝 초기 답변 생성 중...")
        current_answer = self.generate_initial_answer(question)
        
        revisions.append({
            'iteration': 0,
            'answer': current_answer,
            'critique': None
        })
        
        print(f"✅ 초기 답변 완료\n")
        
        # 2. 반복적 개선
        for i in range(self.max_revisions):
            print(f"🔍 수정 {i+1}/{self.max_revisions}: 비평 생성 중...")
            
            # 비평
            critique_result = self.critique_answer(question, current_answer)
            
            if not critique_result['has_issues']:
                print("✅ 더 이상 개선할 점이 없습니다.")
                break
            
            print(f"📋 개선점 발견. 답변 수정 중...")
            
            # 개선
            revised_answer = self.revise_answer(
                question,
                current_answer,
                critique_result['critique']
            )
            
            revisions.append({
                'iteration': i + 1,
                'answer': revised_answer,
                'critique': critique_result['critique']
            })
            
            current_answer = revised_answer
            print(f"✅ 수정 {i+1} 완료\n")
        
        return {
            'method': 'sequential_revision',
            'question': question,
            'revisions': revisions,
            'final_answer': current_answer,
            'num_revisions': len(revisions) - 1
        }

# 테스트
if __name__ == "__main__":
    reasoner = SequentialRevisionReasoner(max_revisions=3)
    
    questions = [
        "인공지능이 사회에 미치는 영향을 긍정적 측면과 부정적 측면에서 설명해주세요.",
        "기후 변화를 완화하기 위한 구체적인 개인 행동 방안을 제시해주세요.",
    ]
    
    for question in questions:
        print(f"\n{'#'*70}")
        print(f"질문: {question}")
        print(f"{'#'*70}\n")
        
        result = reasoner.revise_iteratively(question)
        
        print(f"\n{'='*70}")
        print(f"개선 과정:")
        print(f"{'='*70}\n")
        
        for rev in result['revisions']:
            print(f"--- 반복 {rev['iteration']} ---")
            print(f"답변:\n{rev['answer'][:200]}...\n")
            if rev['critique']:
                print(f"비평:\n{rev['critique'][:200]}...\n")
        
        print(f"{'='*70}")
        print(f"최종 답변 (총 {result['num_revisions']}회 수정):")
        print(f"{'='*70}")
        print(result['final_answer'])

---

🎯 Step 5: 통합 Deep Research 시스템 (30분)

src/deep_research.py

import os
from typing import Dict, List, Optional
from openai import OpenAI
from dotenv import load_dotenv
import sys
sys.path.append('..')

from cot_reasoner import ChainOfThoughtReasoner
from tot_reasoner import TreeOfThoughtsReasoner
from self_consistency import SelfConsistencyReasoner
from sequential_revision import SequentialRevisionReasoner
from tools.search_tools import SearchTools

load_dotenv()

class DeepResearchSystem:
    """통합 Deep Research 시스템"""
    
    def __init__(self, model: str = "gpt-4o"):
        """
        Args:
            model: OpenAI 모델 (gpt-4o 권장)
        """
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.model = model
        
        # 각 추론 모듈 초기화
        self.cot = ChainOfThoughtReasoner(model_type="openai", model=model)
        self.tot = TreeOfThoughtsReasoner(model=model, max_depth=3, branches=3)
        self.consistency = SelfConsistencyReasoner(model=model, num_samples=3)
        self.revision = SequentialRevisionReasoner(model=model, max_revisions=2)
        
        # 검색 도구
        self.search = SearchTools()
    
    def analyze_question_complexity(self, question: str) -> Dict:
        """질문 복잡도 분석"""
        prompt = f"""다음 질문의 복잡도를 분석하세요:

질문: {question}

다음 기준으로 평가하세요 (각 1-5점):
1. 추론 단계 수 (몇 단계의 논리가 필요한가?)
2. 전문 지식 요구도 (얼마나 전문적인 지식이 필요한가?)
3. 모호성 (질문이 얼마나 명확한가?)
4. 탐색 필요성 (여러 방향을 탐색해야 하는가?)

JSON 형식으로 답하세요:
{{"reasoning_steps": 3, "expertise_level": 4, "ambiguity": 2, "exploration_need": 3, "recommended_method": "tree_of_thoughts"}}

recommended_method는 다음 중 하나: simple_cot, tree_of_thoughts, self_consistency, full_research"""

        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0,
            max_tokens=200
        )
        
        import json
        try:
            analysis = json.loads(response.choices[0].message.content)
        except:
            # 파싱 실패 시 기본값
            analysis = {
                "reasoning_steps": 3,
                "expertise_level": 3,
                "ambiguity": 3,
                "exploration_need": 3,
                "recommended_method": "self_consistency"
            }
        
        # 복잡도 점수 계산 (4-20점)
        complexity_score = (
            analysis['reasoning_steps'] +
            analysis['expertise_level'] +
            analysis['ambiguity'] +
            analysis['exploration_need']
        )
        
        analysis['complexity_score'] = complexity_score
        
        return analysis
    
    def gather_context(self, question: str, max_sources: int = 3) -> List[Dict]:
        """외부 정보 수집"""
        print(f"🔍 관련 정보 검색 중...")
        results = self.search.search(question, max_results=max_sources)
        print(f"✅ {len(results)}개 출처 발견\n")
        return results
    
    def deep_research(self, question: str, use_web_search: bool = True) -> Dict:
        """전체 Deep Research 파이프라인"""
        print(f"{'='*70}")
        print(f"🎯 Deep Research 시작")
        print(f"{'='*70}\n")
        print(f"질문: {question}\n")
        
        # 1. 질문 분석
        print("📊 Step 1: 질문 복잡도 분석")
        analysis = self.analyze_question_complexity(question)
        print(f"   복잡도 점수: {analysis['complexity_score']}/20")
        print(f"   권장 방법: {analysis['recommended_method']}\n")
        
        # 2. 컨텍스트 수집 (선택적)
        context_sources = []
        if use_web_search:
            print("📚 Step 2: 외부 정보 수집")
            context_sources = self.gather_context(question)
        
        # 3. 추론 방법 선택 및 실행
        method = analysis['recommended_method']
        complexity = analysis['complexity_score']
        
        if complexity < 8 or method == "simple_cot":
            print("🧠 Step 3: Chain-of-Thought 추론")
            reasoning_result = self.cot.few_shot_cot(question)
            reasoning_method = "CoT"
        
        elif method == "tree_of_thoughts" or complexity >= 15:
            print("🌳 Step 3: Tree of Thoughts 탐색")
            reasoning_result = self.tot.bfs_search(question)
            reasoning_method = "ToT"
        
        elif method == "self_consistency" or 8 <= complexity < 15:
            print("🔄 Step 3: Self-Consistency 추론")
            reasoning_result = self.consistency.reason_with_consistency(question)
            reasoning_method = "Self-Consistency"
        
        else:  # full_research
            print("🔬 Step 3: 전체 연구 모드")
            # ToT + Self-Consistency 조합
            tot_result = self.tot.bfs_search(question)
            consistency_result = self.consistency.reason_with_consistency(question)
            
            reasoning_result = {
                'tot': tot_result,
                'consistency': consistency_result
            }
            reasoning_method = "Full Research"
        
        # 4. 답변 개선
        if reasoning_method != "Full Research":
            print("\n✨ Step 4: 답변 개선 (Sequential Revision)")
            initial_answer = reasoning_result.get('final_answer', reasoning_result.get('reasoning', ''))
            
            # 간단한 개선 (1회만)
            critique = self.revision.critique_answer(question, initial_answer)
            if critique['has_issues']:
                improved_answer = self.revision.revise_answer(
                    question,
                    initial_answer,
                    critique['critique']
                )
            else:
                improved_answer = initial_answer
        else:
            improved_answer = reasoning_result['consistency']['final_answer']
        
        # 5. 최종 종합
        print("\n📝 Step 5: 최종 답변 생성")
        final_answer = self.synthesize_final_answer(
            question,
            improved_answer,
            context_sources
        )
        
        print("\n✅ Deep Research 완료!\n")
        
        return {
            'question': question,
            'analysis': analysis,
            'context_sources': context_sources,
            'reasoning_method': reasoning_method,
            'reasoning_result': reasoning_result,
            'final_answer': final_answer
        }
    
    def synthesize_final_answer(self, question: str, reasoning_answer: str, sources: List[Dict]) -> str:
        """최종 답변 종합"""
        sources_text = ""
        if sources:
            sources_text = "\n\n참고 자료:\n" + "\n".join([
                f"- {s['title']}: {s['content'][:100]}..."
                for s in sources
            ])
        
        prompt = f"""질문: {question}

추론 결과:
{reasoning_answer}
{sources_text}

위 내용을 바탕으로 최종 답변을 작성하세요. 답변은 다음을 포함해야 합니다:
1. 핵심 답변 (간결하고 명확하게)
2. 상세 설명 (필요시)
3. 출처 표시 (외부 자료 사용 시)

답변:"""

        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3,
            max_tokens=1000
        )
        
        return response.choices[0].message.content

# 테스트
if __name__ == "__main__":
    system = DeepResearchSystem(model="gpt-4o-mini")
    
    questions = [
        "기후 변화가 생물 다양성에 미치는 영향과 대응 방안을 설명하세요.",
        "양자컴퓨터가 현대 암호화 시스템에 미칠 위협과 대안을 논의하세요.",
        "24와 36의 최대공약수는?"  # 간단한 질문
    ]
    
    for question in questions:
        result = system.deep_research(question, use_web_search=True)
        
        print(f"\n{'#'*70}")
        print(f"최종 답변:")
        print(f"{'#'*70}")
        print(result['final_answer'])
        print(f"\n사용된 방법: {result['reasoning_method']}")
        print(f"복잡도: {result['analysis']['complexity_score']}/20")

---

🎨 Step 6: Streamlit UI - Deep Research Dashboard (30분)

app.py

import streamlit as st
import sys
sys.path.append('src')
sys.path.append('reasoning')

from deep_research import DeepResearchSystem
from cot_reasoner import ChainOfThoughtReasoner
from tot_reasoner import TreeOfThoughtsReasoner
from self_consistency import SelfConsistencyReasoner
import time

# 페이지 설정
st.set_page_config(
    page_title="Deep Research System",
    page_icon="🔬",
    layout="wide"
)

# CSS
st.markdown("""
<style>
    .main-title {
        font-size: 3rem;
        font-weight: bold;
        text-align: center;
        background: linear-gradient(120deg, #667eea 0%, #764ba2 100%);
        -webkit-background-clip: text;
        -webkit-text-fill-color: transparent;
        margin-bottom: 1rem;
    }
    .complexity-badge {
        display: inline-block;
        padding: 5px 15px;
        border-radius: 20px;
        font-weight: bold;
        margin: 5px;
    }
    .low-complexity { background-color: #d4edda; color: #155724; }
    .medium-complexity { background-color: #fff3cd; color: #856404; }
    .high-complexity { background-color: #f8d7da; color: #721c24; }
    
    .reasoning-step {
        background-color: #f8f9fa;
        border-left: 4px solid #667eea;
        padding: 15px;
        margin: 10px 0;
        border-radius: 5px;
    }
    .final-answer {
        background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
        color: white;
        padding: 25px;
        border-radius: 10px;
        margin: 20px 0;
    }
</style>
""", unsafe_allow_html=True)

# 세션 상태 초기화
if 'system' not in st.session_state:
    st.session_state.system = None
if 'research_history' not in st.session_state:
    st.session_state.research_history = []

# 헤더
st.markdown('<h1 class="main-title">🔬 Deep Research System</h1>', unsafe_allow_html=True)
st.markdown('<p style="text-align: center; font-size: 1.2rem; color: #666;">고급 추론 기법으로 복잡한 질문에 답합니다</p>', unsafe_allow_html=True)

# 사이드바
with st.sidebar:
    st.header("⚙️ 시스템 설정")
    
    model_choice = st.selectbox(
        "LLM 모델",
        ["gpt-4o-mini", "gpt-4o", "gpt-4o-mini"],
        help="gpt-4o가 가장 정확하지만 비용이 높습니다"
    )
    
    reasoning_mode = st.selectbox(
        "추론 모드",
        ["자동 선택 (권장)", "Chain-of-Thought", "Tree of Thoughts", "Self-Consistency", "전체 연구"],
        help="자동 선택은 질문 복잡도에 따라 최적의 방법을 선택합니다"
    )
    
    use_web_search = st.checkbox(
        "웹 검색 활성화",
        value=True,
        help="최신 정보가 필요한 경우 활성화하세요"
    )
    
    show_reasoning_process = st.checkbox(
        "추론 과정 표시",
        value=True,
        help="상세한 추론 단계를 보여줍니다"
    )
    
    # 시스템 초기화
    if st.button("🔄 시스템 초기화"):
        with st.spinner("초기화 중..."):
            st.session_state.system = DeepResearchSystem(model=model_choice)
            st.success("✅ 초기화 완료!")
    
    st.markdown("---")
    
    # 연구 히스토리
    st.subheader("📜 연구 기록")
    if st.session_state.research_history:
        for i, item in enumerate(reversed(st.session_state.research_history[-3:]), 1):
            with st.expander(f"{i}. {item['question'][:30]}..."):
                st.write(f"**방법:** {item['method']}")
                st.write(f"**복잡도:** {item['complexity']}/20")
                st.caption(f"⏱️ {item['time']}")
    else:
        st.info("아직 연구 기록이 없습니다")
    
    if st.button("🗑️ 기록 삭제"):
        st.session_state.research_history = []
        st.rerun()
    
    st.markdown("---")
    
    # 예시 질문
    st.subheader("💡 예시 질문")
    
    st.write("**간단 (CoT)**")
    simple_questions = [
        "24와 36의 최대공약수는?",
        "100달러를 1400원 환율로 환산하면?"
    ]
    for q in simple_questions:
        if st.button(q, key=f"simple_{q}", use_container_width=True):
            st.session_state.current_question = q
    
    st.write("**중간 (Self-Consistency)**")
    medium_questions = [
        "기후 변화의 주요 원인 3가지는?",
        "AI가 일자리에 미치는 영향은?"
    ]
    for q in medium_questions:
        if st.button(q, key=f"medium_{q}", use_container_width=True):
            st.session_state.current_question = q
    
    st.write("**복잡 (ToT/Full)**")
    complex_questions = [
        "양자컴퓨팅이 암호화에 미칠 영향과 대응책을 논하시오",
        "인공지능 윤리 규제의 필요성과 한계를 분석하시오"
    ]
    for q in complex_questions:
        if st.button(q, key=f"complex_{q}", use_container_width=True):
            st.session_state.current_question = q

# 메인 영역
question = st.text_area(
    "🔍 연구 질문 입력",
    height=100,
    placeholder="복잡한 질문을 입력하세요. 예: 양자컴퓨터가 현대 암호화에 미칠 영향을 분석하고 대안을 제시하시오.",
    value=st.session_state.get('current_question', '')
)

col1, col2, col3 = st.columns([1, 1, 3])
with col1:
    research_button = st.button("🚀 연구 시작", type="primary", use_container_width=True)
with col2:
    if st.button("🔄 초기화", use_container_width=True):
        if 'current_question' in st.session_state:
            del st.session_state.current_question
        st.rerun()

if research_button and question:
    # 시스템 초기화 확인
    if st.session_state.system is None:
        with st.spinner("시스템 초기화 중..."):
            st.session_state.system = DeepResearchSystem(model=model_choice)
    
    start_time = time.time()
    
    # 프로그레스 표시
    progress_container = st.container()
    with progress_container:
        progress_bar = st.progress(0)
        status_text = st.empty()
    
    # 연구 실행
    try:
        # Step 1: 질문 분석
        status_text.text("📊 질문 복잡도 분석 중...")
        progress_bar.progress(20)
        
        analysis = st.session_state.system.analyze_question_complexity(question)
        
        # 복잡도 표시
        with st.expander("📊 질문 분석 결과", expanded=True):
            col1, col2, col3, col4 = st.columns(4)
            col1.metric("추론 단계", f"{analysis['reasoning_steps']}/5")
            col2.metric("전문성", f"{analysis['expertise_level']}/5")
            col3.metric("모호성", f"{analysis['ambiguity']}/5")
            col4.metric("탐색 필요", f"{analysis['exploration_need']}/5")
            
            complexity = analysis['complexity_score']
            if complexity < 10:
                badge_class = "low-complexity"
                level = "낮음"
            elif complexity < 15:
                badge_class = "medium-complexity"
                level = "중간"
            else:
                badge_class = "high-complexity"
                level = "높음"
            
            st.markdown(f"""
            <div style='text-align: center; margin-top: 10px;'>
                <span class='complexity-badge {badge_class}'>
                    복잡도: {complexity}/20 ({level})
                </span>
                <span class='complexity-badge' style='background-color: #e7f3ff; color: #004085;'>
                    권장 방법: {analysis['recommended_method']}
                </span>
            </div>
            """, unsafe_allow_html=True)
        
        # Step 2: 웹 검색 (선택적)
        if use_web_search:
            status_text.text("🔍 외부 정보 수집 중...")
            progress_bar.progress(40)
            
            sources = st.session_state.system.gather_context(question)
            
            if sources:
                with st.expander(f"📚 참고 자료 ({len(sources)}개)", expanded=False):
                    for i, source in enumerate(sources, 1):
                        st.markdown(f"""
                        **{i}. [{source['title']}]({source['url']})**  
                        {source['content'][:150]}...  
                        *점수: {source['score']:.2f}*
                        """)
        
        # Step 3: 추론
        status_text.text("🧠 심층 추론 중...")
        progress_bar.progress(60)
        
        if reasoning_mode == "자동 선택 (권장)":
            result = st.session_state.system.deep_research(question, use_web_search)
            method_used = result['reasoning_method']
        elif reasoning_mode == "Chain-of-Thought":
            cot = ChainOfThoughtReasoner(model_type="openai", model=model_choice)
            result = cot.few_shot_cot(question)
            method_used = "CoT"
        elif reasoning_mode == "Tree of Thoughts":
            tot = TreeOfThoughtsReasoner(model=model_choice)
            result = tot.bfs_search(question)
            method_used = "ToT"
        elif reasoning_mode == "Self-Consistency":
            consistency = SelfConsistencyReasoner(model=model_choice, num_samples=3)
            result = consistency.reason_with_consistency(question)
            method_used = "Self-Consistency"
        else:  # 전체 연구
            result = st.session_state.system.deep_research(question, use_web_search)
            method_used = "Full Research"
        
        progress_bar.progress(100)
        status_text.empty()
        progress_bar.empty()
        
        # 추론 과정 표시
        if show_reasoning_process and 'reasoning_result' in result:
            with st.expander("🧠 추론 과정", expanded=False):
                reasoning = result['reasoning_result']
                
                if method_used == "ToT" and 'best_path' in reasoning:
                    st.write("**최적 추론 경로:**")
                    for i, step in enumerate(reasoning['best_path'], 1):
                        st.markdown(f"""
                        <div class="reasoning-step">
                            <strong>Step {i}:</strong> {step}
                        </div>
                        """, unsafe_allow_html=True)
                    st.info(f"점수: {reasoning['best_score']:.2f} | 탐색 노드: {reasoning['total_nodes']}")
                
                elif method_used == "Self-Consistency" and 'paths' in reasoning:
                    st.write(f"**{len(reasoning['paths'])}개의 추론 경로:**")
                    for i, path in enumerate(reasoning['paths'], 1):
                        with st.expander(f"경로 {i}", expanded=False):
                            st.write(path['reasoning'][:300] + "...")
                            st.caption(f"답변: {path['answer']}")
                    
                    st.write("**합의 결과:**")
                    for ans, count in reasoning['consensus']['vote_distribution'].items():
                        st.write(f"- '{ans}': {count}표")
                    st.info(f"신뢰도: {reasoning['consensus']['confidence']:.1%}")
        
        # 최종 답변
        final_answer = result.get('final_answer', result.get('reasoning', '답변 생성 실패'))
        
        st.markdown(f"""
        <div class="final-answer">
            <h3>💡 최종 답변</h3>
            <p style='font-size: 1.1rem; line-height: 1.6;'>{final_answer}</p>
        </div>
        """, unsafe_allow_html=True)
        
        # 메타 정보
        elapsed_time = time.time() - start_time
        col1, col2, col3 = st.columns(3)
        col1.metric("사용된 방법", method_used)
        col2.metric("소요 시간", f"{elapsed_time:.1f}초")
        col3.metric("복잡도", f"{analysis['complexity_score']}/20")
        
        # 히스토리 저장
        st.session_state.research_history.append({
            'question': question,
            'method': method_used,
            'complexity': analysis['complexity_score'],
            'time': time.strftime('%Y-%m-%d %H:%M:%S')
        })
        
        # current_question 초기화
        if 'current_question' in st.session_state:
            del st.session_state.current_question
    
    except Exception as e:
        st.error(f"❌ 오류 발생: {str(e)}")
        st.exception(e)

# Footer
st.markdown("---")
st.markdown("""
<div style='text-align: center; color: #666;'>
    <p><strong>지원되는 추론 기법</strong></p>
    <p>Chain-of-Thought • Tree of Thoughts • Self-Consistency • Sequential Revision</p>
</div>
""", unsafe_allow_html=True)

실행

streamlit run app.py

---

📊 Step 7: 추론 시각화 (추가, 20분)

visualizations/tree_visualizer.py

import networkx as nx
import matplotlib.pyplot as plt
from typing import List, Dict
import sys
sys.path.append('..')

from src.tot_reasoner import TreeNode

class TreeVisualizer:
    """추론 트리 시각화"""
    
    @staticmethod
    def visualize_tot(root_node: TreeNode, filename: str = "reasoning_tree.png"):
        """Tree of Thoughts 시각화"""
        G = nx.DiGraph()
        
        def add_nodes_edges(node, parent_id=None):
            node_id = id(node)
            label = f"{node.state[:30]}...\n점수: {node.value_score:.2f}"
            G.add_node(node_id, label=label, score=node.value_score)
            
            if parent_id is not None:
                G.add_edge(parent_id, node_id)
            
            for child in node.children:
                add_nodes_edges(child, node_id)
        
        add_nodes_edges(root_node)
        
        # 레이아웃
        pos = nx.spring_layout(G, k=2, iterations=50)
        
        # 그리기
        plt.figure(figsize=(15, 10))
        
        # 노드 색상 (점수에 따라)
        node_colors = [G.nodes[node]['score'] for node in G.nodes()]
        
        nx.draw(G, pos,
                labels=nx.get_node_attributes(G, 'label'),
                node_color=node_colors,
                node_size=3000,
                font_size=8,
                font_weight='bold',
                cmap=plt.cm.RdYlGn,
                vmin=0, vmax=1,
                with_labels=True,
                arrows=True,
                edge_color='gray')
        
        plt.title("Tree of Thoughts Visualization", fontsize=16, fontweight='bold')
        plt.tight_layout()
        plt.savefig(filename, dpi=300, bbox_inches='tight')
        print(f"✅ 트리 시각화 저장: {filename}")
        plt.close()

# 테스트
if __name__ == "__main__":
    from src.tot_reasoner import TreeOfThoughtsReasoner
    
    tot = TreeOfThoughtsReasoner(max_depth=2, branches=2)
    question = "12와 18의 최대공약수를 구하세요"
    
    result = tot.bfs_search(question)
    
    # 트리 시각화
    if result['tree_nodes']:
        root = result['tree_nodes'][0]
        TreeVisualizer.visualize_tot(root, "reasoning_tree.png")

---

🎓 학습 과제 및 다음 단계

필수 과제 체크리스트

• [ ] CoT 추론 구현 및 테스트
• [ ] ToT 트리 탐색 동작 확인
• [ ] Self-Consistency 검증
• [ ] Sequential Revision 개선 확인
• [ ] 통합 시스템 테스트

심화 과제

1. 비용 최적화: 토큰 사용량 추적 및 최적화
2. 캐싱: 이전 추론 결과 재사용
3. 병렬화: 여러 추론 경로 동시 실행
4. 사용자 피드백: 답변 평가 및 개선
5. 도메인 특화: 특정 분야(법률, 의료 등)에 최적화