Open Deep Research 완전 가이드 - LangChain AI 에이전트로 구축하는 차세대 연구 자동화 시스템
⏱️ 예상 읽기 시간: 20분
서론: Open Deep Research가 바꾸는 AI 연구의 미래
AI 연구 분야에서 가장 혁신적인 애플리케이션 중 하나인 딥 리서치(Deep Research)가 새로운 전환점을 맞고 있습니다. LangChain AI의 Open Deep Research는 이러한 변화를 주도하는 완전 오픈소스 연구 자동화 에이전트로, 복잡한 연구 과정을 AI가 자동으로 수행할 수 있게 해줍니다.
Open Deep Research의 혁신성
완전한 오픈소스 생태계:
- 🔓 MIT 라이선스: 상업적 사용 포함 완전 자유
- 🌟 커뮤니티 검증: GitHub 6.8k stars, 913 forks의 신뢰성
- 🔧 무제한 커스터마이징: 소스코드 레벨에서 자유로운 수정
- 📚 투명한 개발: 모든 구현 로직 공개
차세대 에이전트 아키텍처:
- 🏗️ LangGraph 기반: 최신 AI 워크플로우 엔진 활용
- 🤖 다중 에이전트: 협업하는 AI 연구원들의 시스템
- 🔄 반성적 사고: 연구 결과를 스스로 검증하고 개선
- ⚡ 병렬 처리: 동시 다발적 연구 수행으로 속도 극대화
범용적 호환성:
- 🎯 다중 모델 지원: OpenAI, Anthropic, Google Vertex AI 등
- 🔍 다양한 검색 API: Tavily, OpenAI Native, Anthropic Native
- 🔌 MCP 서버 연동: Model Context Protocol로 무한 확장
- 📊 포괄적 평가: 다차원 성능 분석 시스템
이 가이드에서 배울 내용
- Open Deep Research 핵심 아키텍처 이해
- 로컬 환경 설정 및 빠른 시작
- LangGraph Studio 활용 방법
- 다중 모델 설정 및 최적화
- MCP 서버 연동 및 확장
- 연구 워크플로우 커스터마이징
- 평가 시스템 구축 및 활용
- 프로덕션 배포 전략
- 실전 연구 프로젝트 구현
Open Deep Research 핵심 아키텍처
전체 시스템 구조
Multi-Agent Research Architecture:
┌─────────────────────────────────────────────────────────────┐
│ Research Supervisor │
│ (연구 총괄 관리 에이전트) │
└─────────────────┬───────────────────────────────────────────┘
│
┌─────────┼─────────┐
│ │ │
┌───────▼─────┐ ┌─▼─────┐ ┌─▼──────────┐
│ Researcher │ │ Search │ │Compression │
│ Agent 1 │ │ Agent │ │ Agent │
└─────────────┘ └────────┘ └────────────┘
│ │
│ ┌─────────────┐ │
└────► Research ◄──────┘
│ Results │
│ Database │
└─────────────┘
주요 구성 요소 분석
1. Research Supervisor (연구 감독자)
- 역할: 전체 연구 프로세스 조율 및 관리
- 기능: 연구 계획 수립, 하위 에이전트 관리, 품질 검증
- 모델: GPT-4o 등 고성능 모델 권장
2. Sub-Research Agents (연구 에이전트들)
- 역할: 특정 주제에 대한 심화 연구 수행
- 기능: 정보 수집, 분석, 중간 보고서 작성
- 특징: 병렬 실행으로 연구 속도 향상
3. Search Integration Layer
- Tavily API: 모든 모델과 호환되는 범용 검색
- Native Search: OpenAI/Anthropic 전용 고급 검색
- Custom Search: 사용자 정의 검색 엔진 연동
4. MCP (Model Context Protocol) System
- 로컬 MCP: 파일시스템, 데이터베이스 접근
- 원격 MCP: 클라우드 서비스, API 연동
- 확장성: 무제한 도구 및 서비스 연결
빠른 시작 가이드 (Quickstart)
환경 준비 및 설치
1단계: 시스템 요구사항 확인
# Python 3.11+ 필수
python --version
# uv 패키지 매니저 설치 (권장)
curl -LsSf https://astral.sh/uv/install.sh | sh
# 또는 pip 설치
pip install uv
2단계: 프로젝트 클론 및 환경 설정
# GitHub에서 프로젝트 클론
git clone https://github.com/langchain-ai/open_deep_research.git
cd open_deep_research
# 가상환경 생성 및 활성화
uv venv
source .venv/bin/activate # macOS/Linux
# .venv\Scripts\activate # Windows
# 의존성 설치
uv pip install -r pyproject.toml
3단계: 환경 변수 설정
# 환경 변수 파일 생성
cp .env.example .env
# .env 파일 편집
nano .env
필수 환경 변수 설정:
# OpenAI API 설정
OPENAI_API_KEY=sk-your-openai-api-key
# Tavily Search API (무료 계정으로 시작 가능)
TAVILY_API_KEY=tvly-your-tavily-api-key
# LangSmith 추적 (선택적)
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=ls__your-langsmith-api-key
LANGCHAIN_PROJECT=open-deep-research
# Anthropic API (선택적)
ANTHROPIC_API_KEY=sk-ant-your-anthropic-key
# Google Vertex AI (선택적)
GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account.json
LangGraph Studio 실행
4단계: 개발 서버 시작
# LangGraph 개발 서버 실행
uvx --refresh --from "langgraph-cli[inmem]" --with-editable . --python 3.11 langgraph dev --allow-blocking
성공적 실행 확인:
🚀 API: http://127.0.0.1:2024
🎨 Studio UI: https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024
📚 API Docs: http://127.0.0.1:2024/docs
5단계: LangGraph Studio 사용
- Studio UI 접속: https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024
- Graph 선택: open_deep_research 그래프 선택
- 첫 연구 실행:
messages입력 필드에 연구 질문 입력- 예시: “Explain the latest developments in quantum computing”
Submit버튼 클릭
첫 연구 프로젝트 실행
간단한 연구 질문 예제:
# examples/simple_research.py
from src.research_assistant import research_assistant
async def run_simple_research():
# 연구 질문 정의
question = "What are the latest breakthroughs in AI safety research in 2024?"
# 연구 실행
result = await research_assistant.ainvoke({
"messages": [{"role": "user", "content": question}]
})
# 결과 출력
print("연구 결과:")
print(result["messages"][-1]["content"])
# 실행
import asyncio
asyncio.run(run_simple_research())
실행 결과 예제:
연구 결과:
# AI Safety Research: 2024년 주요 돌파구
## 1. Constitutional AI 발전
- Anthropic의 Claude 3.5 Sonnet에서 개선된 헌법적 AI 구현
- 더 나은 가치 정렬과 해악 방지 메커니즘
## 2. Scalable Oversight
- OpenAI의 슈퍼정렬 팀 연구 결과
- 인간보다 뛰어난 AI 시스템의 감독 방법론
## 3. Interpretability Tools
- Anthropic의 Dictionary Learning for SAEs
- 신경망 내부 표현의 해석 가능성 향상
[상세한 연구 결과 계속...]
고급 설정 및 구성 옵션
연구 동작 설정
핵심 설정 파라미터:
# src/config/research_config.py
from dataclasses import dataclass
from typing import Optional
@dataclass
class ResearchConfig:
# 일반 설정
max_structured_output_retries: int = 3
allow_clarification: bool = True
max_concurrent_research_units: int = 5
# 연구 프로세스 설정
search_api: str = "tavily" # tavily, openai_native, anthropic_native
max_researcher_iterations: int = 3
max_react_tool_calls: int = 5
# 모델 설정
summarization_model: str = "openai:gpt-4o-mini"
research_model: str = "openai:gpt-4o"
compression_model: str = "openai:gpt-4o-mini"
final_report_model: str = "openai:gpt-4o"
모델 선택 및 최적화
모델별 특성 및 권장 사용:
# src/models/model_selector.py
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_vertexai import ChatVertexAI
class ModelSelector:
def __init__(self):
self.model_configs = {
# OpenAI 모델들
"openai:gpt-4o": {
"class": ChatOpenAI,
"model": "gpt-4o",
"temperature": 0.1,
"max_tokens": 4000,
"supports_structured_output": True,
"cost_per_1k_tokens": 0.03,
"best_for": ["research", "final_report"]
},
"openai:gpt-4o-mini": {
"class": ChatOpenAI,
"model": "gpt-4o-mini",
"temperature": 0.1,
"max_tokens": 2000,
"supports_structured_output": True,
"cost_per_1k_tokens": 0.005,
"best_for": ["summarization", "compression"]
},
# Anthropic 모델들
"anthropic:claude-3-5-sonnet": {
"class": ChatAnthropic,
"model": "claude-3-5-sonnet-20241022",
"temperature": 0.1,
"max_tokens": 4000,
"supports_structured_output": True,
"cost_per_1k_tokens": 0.03,
"best_for": ["research", "analysis"]
},
# Google 모델들
"vertex:gemini-pro": {
"class": ChatVertexAI,
"model": "gemini-pro",
"temperature": 0.1,
"max_output_tokens": 2048,
"supports_structured_output": True,
"cost_per_1k_tokens": 0.0025,
"best_for": ["summarization", "compression"]
}
}
def get_optimized_model(self, task_type: str, budget_level: str = "standard"):
"""태스크 유형과 예산에 따른 최적 모델 선택"""
if budget_level == "premium":
if task_type in ["research", "final_report"]:
return "anthropic:claude-3-5-sonnet"
else:
return "openai:gpt-4o-mini"
elif budget_level == "budget":
return "openai:gpt-4o-mini"
else: # standard
if task_type in ["research", "final_report"]:
return "openai:gpt-4o"
else:
return "openai:gpt-4o-mini"
검색 API 설정 및 비교
다양한 검색 옵션 설정:
# src/search/search_manager.py
from typing import Dict, Any
import os
class SearchManager:
def __init__(self):
self.search_configs = {
"tavily": {
"api_key": os.getenv("TAVILY_API_KEY"),
"max_results": 10,
"search_depth": "advanced",
"include_images": False,
"include_answer": True,
"compatibility": "all_models",
"cost": "low"
},
"openai_native": {
"requires_model": "openai",
"web_search_enabled": True,
"real_time_data": True,
"accuracy": "high",
"cost": "medium"
},
"anthropic_native": {
"requires_model": "anthropic",
"web_search_enabled": True,
"real_time_data": True,
"accuracy": "high",
"cost": "medium"
}
}
def configure_search(self, search_type: str, custom_config: Dict[str, Any] = None):
"""검색 엔진 설정"""
base_config = self.search_configs.get(search_type, {})
if custom_config:
base_config.update(custom_config)
return base_config
# 사용 예제
search_manager = SearchManager()
# Tavily 설정 (가장 범용적)
tavily_config = search_manager.configure_search("tavily", {
"max_results": 15,
"search_depth": "advanced",
"include_domain_filter": ["arxiv.org", "nature.com", "science.org"]
})
MCP (Model Context Protocol) 서버 연동
로컬 MCP 서버 설정
파일시스템 MCP 서버 구성:
# src/mcp/filesystem_mcp.py
import json
import subprocess
from pathlib import Path
class FilesystemMCP:
def __init__(self, allowed_dirs: list):
self.allowed_dirs = [Path(d).resolve() for d in allowed_dirs]
self.server_process = None
def start_server(self):
"""파일시스템 MCP 서버 시작"""
cmd = [
"mcp-server-filesystem",
*[str(d) for d in self.allowed_dirs]
]
self.server_process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
return self.server_process
def configure_security(self):
"""보안 설정 구성"""
return {
"allowed_operations": [
"read_file",
"write_file",
"list_directory",
"create_directory",
"search_files"
],
"restricted_patterns": [
"*.exe",
"*.sh",
"/etc/*",
"/sys/*"
],
"max_file_size": "10MB",
"max_depth": 10
}
# 사용 예제
filesystem_mcp = FilesystemMCP([
"/Users/researcher/documents",
"/Users/researcher/projects",
"/tmp/research_workspace"
])
security_config = filesystem_mcp.configure_security()
server = filesystem_mcp.start_server()
SQLite 데이터베이스 MCP 연동:
# src/mcp/database_mcp.py
import sqlite3
import json
from typing import Dict, Any
class DatabaseMCP:
def __init__(self, db_path: str):
self.db_path = db_path
self.connection = None
def setup_research_db(self):
"""연구 데이터베이스 스키마 설정"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# 연구 프로젝트 테이블
cursor.execute("""
CREATE TABLE IF NOT EXISTS research_projects (
id INTEGER PRIMARY KEY,
title TEXT NOT NULL,
description TEXT,
status TEXT DEFAULT 'active',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# 연구 결과 테이블
cursor.execute("""
CREATE TABLE IF NOT EXISTS research_results (
id INTEGER PRIMARY KEY,
project_id INTEGER,
query TEXT NOT NULL,
result_text TEXT,
sources JSON,
confidence_score REAL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (project_id) REFERENCES research_projects (id)
)
""")
# 인덱스 생성
cursor.execute("CREATE INDEX IF NOT EXISTS idx_results_project ON research_results(project_id)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_results_created ON research_results(created_at)")
conn.commit()
conn.close()
def save_research_result(self, project_id: int, query: str, result: Dict[str, Any]):
"""연구 결과 저장"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
INSERT INTO research_results
(project_id, query, result_text, sources, confidence_score)
VALUES (?, ?, ?, ?, ?)
""", (
project_id,
query,
result.get('text', ''),
json.dumps(result.get('sources', [])),
result.get('confidence', 0.0)
))
conn.commit()
conn.close()
# 사용 예제
db_mcp = DatabaseMCP("/path/to/research.db")
db_mcp.setup_research_db()
원격 MCP 서버 연동
Arcade MCP 서버 설정 (여행 관련 연구):
# src/mcp/remote_mcp.py
import httpx
import json
from typing import Dict, Any
class RemoteMCP:
def __init__(self, server_config: Dict[str, Any]):
self.url = server_config["url"]
self.tools = server_config.get("tools", [])
self.auth_token = server_config.get("auth_token")
self.client = httpx.AsyncClient()
async def setup_arcade_travel_research(self):
"""Arcade 여행 연구 MCP 설정"""
config = {
"url": "https://api.arcade.dev/v1/mcps/ms_0ujssxh0cECutqzMgbtXSGnjorm",
"tools": [
"Search_SearchHotels",
"Search_SearchOneWayFlights",
"Search_SearchRoundtripFlights"
],
"auth_required": True
}
return config
async def call_mcp_tool(self, tool_name: str, parameters: Dict[str, Any]):
"""원격 MCP 도구 호출"""
headers = {"Content-Type": "application/json"}
if self.auth_token:
headers["Authorization"] = f"Bearer {self.auth_token}"
payload = {
"tool": tool_name,
"parameters": parameters
}
response = await self.client.post(
f"{self.url}/call_tool",
json=payload,
headers=headers
)
return response.json()
# 실제 사용 예제
async def research_travel_options():
"""여행 옵션 연구 예제"""
arcade_config = {
"url": "https://api.arcade.dev/v1/mcps/ms_0ujssxh0cECutqzMgbtXSGnjorm",
"tools": ["Search_SearchHotels", "Search_SearchRoundtripFlights"]
}
remote_mcp = RemoteMCP(arcade_config)
# 호텔 검색
hotel_results = await remote_mcp.call_mcp_tool("Search_SearchHotels", {
"destination": "Tokyo",
"check_in": "2024-03-15",
"check_out": "2024-03-20",
"guests": 2
})
# 항공편 검색
flight_results = await remote_mcp.call_mcp_tool("Search_SearchRoundtripFlights", {
"origin": "JFK",
"destination": "NRT",
"departure_date": "2024-03-15",
"return_date": "2024-03-20",
"passengers": 2
})
return {
"hotels": hotel_results,
"flights": flight_results
}
평가 시스템 구축 및 활용
다차원 평가 프레임워크
평가 지표 설계:
# tests/evaluators.py
from typing import Dict, Any, List
from dataclasses import dataclass
@dataclass
class EvaluationResult:
accuracy: float # 0-1 scale
completeness: float # 0-1 scale
relevance: float # 0-1 scale
clarity: float # 0-1 scale
source_quality: float # 0-1 scale
overall_score: float # weighted average
feedback: str
class ResearchEvaluator:
def __init__(self):
self.weights = {
"accuracy": 0.3,
"completeness": 0.25,
"relevance": 0.2,
"clarity": 0.15,
"source_quality": 0.1
}
def evaluate_research_output(self,
research_result: str,
reference_answer: str,
sources: List[Dict[str, Any]]) -> EvaluationResult:
"""연구 결과 종합 평가"""
# 정확성 평가
accuracy = self._evaluate_accuracy(research_result, reference_answer)
# 완성도 평가
completeness = self._evaluate_completeness(research_result, reference_answer)
# 관련성 평가
relevance = self._evaluate_relevance(research_result, reference_answer)
# 명확성 평가
clarity = self._evaluate_clarity(research_result)
# 소스 품질 평가
source_quality = self._evaluate_source_quality(sources)
# 전체 점수 계산
overall_score = (
accuracy * self.weights["accuracy"] +
completeness * self.weights["completeness"] +
relevance * self.weights["relevance"] +
clarity * self.weights["clarity"] +
source_quality * self.weights["source_quality"]
)
# 피드백 생성
feedback = self._generate_feedback(accuracy, completeness, relevance, clarity, source_quality)
return EvaluationResult(
accuracy=accuracy,
completeness=completeness,
relevance=relevance,
clarity=clarity,
source_quality=source_quality,
overall_score=overall_score,
feedback=feedback
)
def _evaluate_accuracy(self, result: str, reference: str) -> float:
"""정확성 평가 - 사실 확인"""
# LLM 기반 팩트 체킹
prompt = f"""
Research Result: {result}
Reference Answer: {reference}
Rate the factual accuracy of the research result compared to the reference on a scale of 0.0 to 1.0.
Consider:
- Factual correctness
- No misinformation
- Proper context
Return only the numerical score.
"""
# 실제 구현에서는 LLM 호출
return 0.85 # 예시 점수
def _evaluate_completeness(self, result: str, reference: str) -> float:
"""완성도 평가 - 필요한 정보 포함 여부"""
prompt = f"""
Research Result: {result}
Reference Answer: {reference}
Rate how completely the research result covers the topic compared to the reference on a scale of 0.0 to 1.0.
Consider:
- Coverage of key points
- Depth of information
- Missing important aspects
Return only the numerical score.
"""
return 0.78 # 예시 점수
배치 평가 시스템
대규모 테스트 데이터셋 평가:
# tests/run_evaluate.py
import asyncio
import json
from typing import List, Dict
from pathlib import Path
class BatchEvaluator:
def __init__(self, evaluator: ResearchEvaluator):
self.evaluator = evaluator
self.results = []
async def run_batch_evaluation(self, test_dataset: List[Dict[str, Any]]):
"""배치 평가 실행"""
print(f"Starting batch evaluation with {len(test_dataset)} test cases...")
for i, test_case in enumerate(test_dataset):
print(f"Processing test case {i+1}/{len(test_dataset)}: {test_case['question'][:50]}...")
# 연구 실행
research_result = await self._run_research(test_case["question"])
# 평가 수행
evaluation = self.evaluator.evaluate_research_output(
research_result["content"],
test_case["reference_answer"],
research_result["sources"]
)
# 결과 저장
result = {
"test_case_id": test_case["id"],
"question": test_case["question"],
"research_result": research_result["content"],
"evaluation": evaluation.__dict__,
"execution_time": research_result["execution_time"]
}
self.results.append(result)
print(f" Overall Score: {evaluation.overall_score:.3f}")
return self.results
def generate_evaluation_report(self) -> Dict[str, Any]:
"""평가 리포트 생성"""
if not self.results:
return {"error": "No evaluation results available"}
# 통계 계산
scores = [r["evaluation"]["overall_score"] for r in self.results]
report = {
"summary": {
"total_cases": len(self.results),
"average_score": sum(scores) / len(scores),
"max_score": max(scores),
"min_score": min(scores),
"median_score": sorted(scores)[len(scores)//2]
},
"dimension_breakdown": {
"accuracy": sum(r["evaluation"]["accuracy"] for r in self.results) / len(self.results),
"completeness": sum(r["evaluation"]["completeness"] for r in self.results) / len(self.results),
"relevance": sum(r["evaluation"]["relevance"] for r in self.results) / len(self.results),
"clarity": sum(r["evaluation"]["clarity"] for r in self.results) / len(self.results),
"source_quality": sum(r["evaluation"]["source_quality"] for r in self.results) / len(self.results)
},
"performance_analysis": {
"avg_execution_time": sum(r["execution_time"] for r in self.results) / len(self.results),
"cases_above_threshold": len([s for s in scores if s >= 0.8]),
"improvement_areas": self._identify_improvement_areas()
},
"detailed_results": self.results
}
return report
def _identify_improvement_areas(self) -> List[str]:
"""개선 영역 식별"""
dimension_scores = {
"accuracy": sum(r["evaluation"]["accuracy"] for r in self.results) / len(self.results),
"completeness": sum(r["evaluation"]["completeness"] for r in self.results) / len(self.results),
"relevance": sum(r["evaluation"]["relevance"] for r in self.results) / len(self.results),
"clarity": sum(r["evaluation"]["clarity"] for r in self.results) / len(self.results),
"source_quality": sum(r["evaluation"]["source_quality"] for r in self.results) / len(self.results)
}
# 평균 이하 성능 영역 식별
avg_score = sum(dimension_scores.values()) / len(dimension_scores)
improvement_areas = [
dimension for dimension, score in dimension_scores.items()
if score < avg_score - 0.05
]
return improvement_areas
# 사용 예제
async def run_comprehensive_evaluation():
"""종합 평가 실행"""
# 테스트 데이터셋 로드
test_dataset = [
{
"id": "ai_safety_001",
"question": "What are the latest developments in AI safety research?",
"reference_answer": "Recent developments include constitutional AI, scalable oversight, and interpretability tools..."
},
{
"id": "quantum_001",
"question": "Explain quantum error correction breakthroughs in 2024",
"reference_answer": "Major breakthroughs include surface code improvements, logical qubit demonstrations..."
}
# ... 더 많은 테스트 케이스
]
# 평가 실행
evaluator = ResearchEvaluator()
batch_evaluator = BatchEvaluator(evaluator)
results = await batch_evaluator.run_batch_evaluation(test_dataset)
report = batch_evaluator.generate_evaluation_report()
# 결과 저장
with open("evaluation_report.json", "w") as f:
json.dump(report, f, indent=2)
print(f"Evaluation completed. Average score: {report['summary']['average_score']:.3f}")
return report
# 실행
# asyncio.run(run_comprehensive_evaluation())
프로덕션 배포 및 확장 전략
Docker 컨테이너화
최적화된 Dockerfile:
# Dockerfile
FROM python:3.11-slim
# 시스템 의존성 설치
RUN apt-get update && apt-get install -y \
git \
curl \
&& rm -rf /var/lib/apt/lists/*
# uv 설치
RUN pip install uv
# 작업 디렉토리 설정
WORKDIR /app
# 의존성 파일 복사
COPY pyproject.toml uv.lock ./
# 의존성 설치
RUN uv pip install --system -r pyproject.toml
# 애플리케이션 코드 복사
COPY . .
# 환경 변수 설정
ENV PYTHONPATH=/app
ENV LANGCHAIN_TRACING_V2=true
# 포트 노출
EXPOSE 8000
# 헬스체크 추가
HEALTHCHECK --interval=30s --timeout=30s --start-period=5s --retries=3 \
CMD curl -f http://localhost:8000/health || exit 1
# 애플리케이션 실행
CMD ["uvx", "--from", "langgraph-cli[inmem]", "--with-editable", ".", "langgraph", "start", "--host", "0.0.0.0", "--port", "8000"]
Docker Compose 설정:
# docker-compose.yml
version: '3.8'
services:
open-deep-research:
build: .
ports:
- "8000:8000"
environment:
- OPENAI_API_KEY=${OPENAI_API_KEY}
- TAVILY_API_KEY=${TAVILY_API_KEY}
- LANGCHAIN_API_KEY=${LANGCHAIN_API_KEY}
- LANGCHAIN_PROJECT=open-deep-research-prod
- REDIS_URL=redis://redis:6379
depends_on:
- redis
- postgres
volumes:
- ./data:/app/data
- ./logs:/app/logs
restart: unless-stopped
redis:
image: redis:7-alpine
ports:
- "6379:6379"
volumes:
- redis_data:/data
command: redis-server --appendonly yes
restart: unless-stopped
postgres:
image: postgres:15
ports:
- "5432:5432"
environment:
- POSTGRES_DB=research_db
- POSTGRES_USER=researcher
- POSTGRES_PASSWORD=${POSTGRES_PASSWORD}
volumes:
- postgres_data:/var/lib/postgresql/data
- ./init.sql:/docker-entrypoint-initdb.d/init.sql
restart: unless-stopped
nginx:
image: nginx:alpine
ports:
- "80:80"
- "443:443"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf
- ./ssl:/etc/nginx/ssl
depends_on:
- open-deep-research
restart: unless-stopped
volumes:
redis_data:
postgres_data:
Kubernetes 배포
Kubernetes 매니페스트:
# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: open-deep-research
labels:
app: open-deep-research
spec:
replicas: 3
selector:
matchLabels:
app: open-deep-research
template:
metadata:
labels:
app: open-deep-research
spec:
containers:
- name: open-deep-research
image: your-registry/open-deep-research:latest
ports:
- containerPort: 8000
env:
- name: OPENAI_API_KEY
valueFrom:
secretKeyRef:
name: api-secrets
key: openai-api-key
- name: TAVILY_API_KEY
valueFrom:
secretKeyRef:
name: api-secrets
key: tavily-api-key
resources:
requests:
memory: "512Mi"
cpu: "250m"
limits:
memory: "2Gi"
cpu: "1000m"
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 30
periodSeconds: 10
readinessProbe:
httpGet:
path: /ready
port: 8000
initialDelaySeconds: 5
periodSeconds: 5
---
apiVersion: v1
kind: Service
metadata:
name: open-deep-research-service
spec:
selector:
app: open-deep-research
ports:
- protocol: TCP
port: 80
targetPort: 8000
type: ClusterIP
---
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: open-deep-research-ingress
annotations:
kubernetes.io/ingress.class: nginx
cert-manager.io/cluster-issuer: letsencrypt-prod
spec:
tls:
- hosts:
- research-api.yourdomain.com
secretName: research-api-tls
rules:
- host: research-api.yourdomain.com
http:
paths:
- path: /
pathType: Prefix
backend:
service:
name: open-deep-research-service
port:
number: 80
실전 연구 프로젝트 구현
복합 연구 워크플로우
다단계 연구 프로세스 설계:
# src/workflows/complex_research.py
from typing import List, Dict, Any
from dataclasses import dataclass
import asyncio
@dataclass
class ResearchPhase:
name: str
description: str
agents_required: int
max_duration: int # seconds
dependencies: List[str]
class ComplexResearchWorkflow:
def __init__(self):
self.phases = {
"literature_review": ResearchPhase(
name="Literature Review",
description="Comprehensive review of existing literature",
agents_required=2,
max_duration=300,
dependencies=[]
),
"data_collection": ResearchPhase(
name="Data Collection",
description="Gather relevant data and statistics",
agents_required=3,
max_duration=600,
dependencies=["literature_review"]
),
"analysis": ResearchPhase(
name="Analysis",
description="Deep analysis of collected information",
agents_required=2,
max_duration=400,
dependencies=["literature_review", "data_collection"]
),
"synthesis": ResearchPhase(
name="Synthesis",
description="Synthesize findings into coherent insights",
agents_required=1,
max_duration=300,
dependencies=["analysis"]
),
"validation": ResearchPhase(
name="Validation",
description="Validate findings and check for consistency",
agents_required=1,
max_duration=200,
dependencies=["synthesis"]
)
}
async def execute_research_pipeline(self, research_question: str) -> Dict[str, Any]:
"""복합 연구 파이프라인 실행"""
results = {}
execution_order = self._determine_execution_order()
print(f"Starting complex research pipeline for: {research_question}")
for phase_name in execution_order:
phase = self.phases[phase_name]
print(f"\n🔬 Executing Phase: {phase.name}")
print(f" Description: {phase.description}")
print(f" Agents: {phase.agents_required}, Max Duration: {phase.max_duration}s")
# 의존성 확인
dependencies_met = all(dep in results for dep in phase.dependencies)
if not dependencies_met:
missing = [dep for dep in phase.dependencies if dep not in results]
raise Exception(f"Dependencies not met for {phase_name}: {missing}")
# 단계별 연구 실행
phase_result = await self._execute_phase(
phase_name,
research_question,
results,
phase
)
results[phase_name] = phase_result
print(f" ✅ Phase completed: {len(phase_result.get('findings', []))} findings")
# 최종 리포트 생성
final_report = await self._generate_final_report(research_question, results)
return {
"research_question": research_question,
"phase_results": results,
"final_report": final_report,
"execution_summary": self._create_execution_summary(results)
}
async def _execute_phase(self,
phase_name: str,
research_question: str,
previous_results: Dict[str, Any],
phase: ResearchPhase) -> Dict[str, Any]:
"""개별 연구 단계 실행"""
# 단계별 맞춤형 프롬프트 생성
phase_prompt = self._generate_phase_prompt(
phase_name,
research_question,
previous_results
)
# 다중 에이전트 병렬 실행
agent_tasks = []
for i in range(phase.agents_required):
task = self._run_phase_agent(
f"agent_{i}",
phase_prompt,
research_question,
previous_results
)
agent_tasks.append(task)
# 결과 수집 및 통합
agent_results = await asyncio.gather(*agent_tasks)
# 결과 통합 및 정제
integrated_result = await self._integrate_agent_results(
phase_name,
agent_results
)
return integrated_result
def _generate_phase_prompt(self,
phase_name: str,
research_question: str,
previous_results: Dict[str, Any]) -> str:
"""단계별 맞춤형 프롬프트 생성"""
phase_prompts = {
"literature_review": f"""
Conduct a comprehensive literature review for the research question: "{research_question}"
Focus on:
1. Recent academic papers and publications
2. Established theories and frameworks
3. Key researchers and institutions
4. Research gaps and controversies
Provide citations and assess credibility of sources.
""",
"data_collection": f"""
Collect relevant data and statistics for: "{research_question}"
Previous literature review findings: {previous_results.get('literature_review', {}).get('summary', 'None')}
Focus on:
1. Quantitative data and statistics
2. Recent surveys and reports
3. Government and institutional data
4. Industry reports and market research
Ensure data is current and from reliable sources.
""",
"analysis": f"""
Conduct deep analysis based on collected information for: "{research_question}"
Previous findings:
- Literature: {previous_results.get('literature_review', {}).get('summary', 'None')}
- Data: {previous_results.get('data_collection', {}).get('summary', 'None')}
Focus on:
1. Pattern identification and trends
2. Cause-effect relationships
3. Comparative analysis
4. Critical evaluation of findings
""",
"synthesis": f"""
Synthesize all research findings into coherent insights for: "{research_question}"
Integration of:
- Literature insights: {previous_results.get('literature_review', {}).get('key_findings', [])}
- Data insights: {previous_results.get('data_collection', {}).get('key_findings', [])}
- Analysis results: {previous_results.get('analysis', {}).get('key_findings', [])}
Create unified conclusions and actionable insights.
""",
"validation": f"""
Validate and verify the research findings for: "{research_question}"
Synthesized insights: {previous_results.get('synthesis', {}).get('summary', 'None')}
Focus on:
1. Fact-checking key claims
2. Consistency verification
3. Bias detection
4. Reliability assessment
"""
}
return phase_prompts.get(phase_name, f"Research the topic: {research_question}")
# 실제 사용 예제
async def run_complex_ai_safety_research():
"""AI 안전성에 대한 복합 연구 실행"""
workflow = ComplexResearchWorkflow()
research_question = """
What are the most critical challenges in AI safety for large language models,
and what are the most promising approaches to address them in 2024?
"""
result = await workflow.execute_research_pipeline(research_question)
print("\n" + "="*80)
print("🎯 FINAL RESEARCH REPORT")
print("="*80)
print(result["final_report"])
print("\n" + "="*80)
print("📊 EXECUTION SUMMARY")
print("="*80)
for phase, summary in result["execution_summary"].items():
print(f"{phase}: {summary}")
return result
# 실행
# asyncio.run(run_complex_ai_safety_research())
결론: Open Deep Research의 무한한 가능성
Open Deep Research가 제공하는 혁신
연구 패러다임의 전환:
- 🤖 AI 협업 연구: 인간과 AI가 함께하는 새로운 연구 방식
- ⚡ 속도의 혁명: 몇 주 걸리던 연구를 몇 시간으로 단축
- 🎯 정확성 향상: 다중 검증과 반성적 사고로 더 높은 신뢰성
- 🌐 접근성 확대: 고급 연구 도구의 민주화
실무 적용 가능성:
- 📊 시장 조사: 경쟁사 분석, 트렌드 예측, 소비자 행동 연구
- 🔬 학술 연구: 문헌 리뷰, 데이터 분석, 가설 검증
- 💼 비즈니스 전략: 산업 분석, 리스크 평가, 기회 탐색
- 🏛️ 정책 연구: 사회 문제 분석, 정책 효과 예측
다음 단계 및 발전 방향
즉시 활용 가능한 프로젝트:
- 개인 연구 어시스턴트: 일상적인 질문과 조사 자동화
- 업무 리서치 도구: 회사 내 시장 조사 및 경쟁 분석
- 학습 도우미: 새로운 기술이나 도메인 학습 지원
- 콘텐츠 제작: 블로그, 리포트, 프레젠테이션 자료 생성
고급 확장 방향:
- 도메인 특화: 의료, 법률, 금융 등 전문 분야 맞춤화
- 실시간 모니터링: 지속적인 정보 추적과 알림 시스템
- 협업 플랫폼: 팀 단위 연구 프로젝트 관리
- 지식 베이스: 조직 내 축적된 연구 자산 활용
커뮤니티와 기여
오픈소스 생태계 참여:
- 🛠️ GitHub 기여: 프로젝트 개선 및 버그 리포트
- 💡 아이디어 제안: 새로운 기능 및 사용 사례 공유
- 📖 문서화: 사용 가이드 및 튜토리얼 작성
- 🎓 교육 콘텐츠: 워크샵, 강의, 블로그 포스트 제작
지속적인 학습과 발전:
- 📚 새로운 모델과 API 출시 동향 파악
- 🔄 워크플로우 최적화 및 성능 튜닝
- 📊 평가 지표 개선 및 벤치마크 구축
- 🌟 실제 사용 사례 공유 및 베스트 프랙티스 개발
Open Deep Research는 단순한 도구를 넘어서 연구와 지식 탐구의 새로운 가능성을 열어주는 플랫폼입니다. 이 가이드를 통해 구축한 시스템을 바탕으로 더욱 혁신적이고 효과적인 연구 프로젝트를 진행하시기 바랍니다.
🚀 지금 바로 시작하세요: GitHub에서 프로젝트를 클론하고 첫 번째 AI 연구 에이전트를 구축해보세요!
“미래의 연구는 AI와 함께합니다. Open Deep Research로 그 미래를 지금 경험하세요.”