웹 및 로컬 문서에서 인용된 보고서를 작성하는 심층 연구 에이전트
README 링크에서 발견 · today
계획을 세우고, 검색하고, 인용이 포함된 보고서를 작성하는 리서치 에이전트를 만들었습니다.
What they did
이들은 주어진 작업에 대해 웹과 로컬 리서치를 모두 처리하는 오픈 딥 리서치 에이전트인 GPT Researcher 위에 딥 리서치 워크플로를 구축했습니다. 이 에이전트는 리서치 브리프를 받아 하위 질문으로 분해하고, 웹 소스와 로컬 문서 전반에 걸쳐 병렬 검색을 실행한 뒤, 그 결과를 인용이 포함된 상세하고 사실에 기반하며 편향되지 않은 보고서로 종합합니다. 이들은 AG2를 오케스트레이션 계층으로 사용해 플래너, 검색 단계, 작성자를 조율함으로써 파이프라인을 추가 에이전트나 커스텀 단계로 확장할 수 있게 했습니다. 출력물은 채팅 답변이 아니라 인라인 인용이 포함된 장문 보고서였기에, 내부 브리프, 경쟁사 스캔, 문헌 스타일 요약의 초안으로 활용할 수 있었습니다. 이들은 또한 프로젝트의 커스터마이제이션 옵션을 활용해 보고서 구조와 소스 처리를 조정했으며, Python 패키지와 제공된 Docker/Colab 경로 양쪽에서 실행해 로컬 실행과 호스팅 실행을 비교했습니다.
How it went
파이프라인은 탄탄한 초안처럼 읽히는, 인용이 뒷받침된 구조화된 보고서를 생성하여 리서치 작업의 백지 단계를 줄여주었다. 병렬 하위 질문 검색은 다중 소스 브리프에서도 지연 시간을 합리적으로 유지했고, AG2 레이어 덕분에 단계 추가가 간단했다. 완성도 수준은 프로토타입에서 내부 도구 정도: 외부로 내보내기 전에 소스 품질과 주장 정확성을 위해 사람이 한 번 검토해야 한다.
Worth knowing
보고서 품질은 기반 모델과 검색 제공자에 크게 좌우되며, 인용이 약하거나 주변적인 출처를 가리킬 수 있어 수동 검증이 필요합니다. 긴 브리프는 실제 토큰과 시간을 소모하므로 예산과 속도 제한이 중요합니다. 로컬 문서 리서치는 자체 문서 저장소와 임베딩을 연결해야 하며, README 발췌에서는 해당 설정을 자세히 다루지 않습니다. 내장된 팩트체크 단계가 없으므로 출력물은 초안으로 취급하세요.
Wiring
gpt-researcher 패키지를 설치하거나(또는 저장소를 클론) 하고, 선택한 LLM과 검색 제공자에 대한 API 키를 .env 파일에 설정합니다. AG2는 그 위에 오케스트레이션 계층으로 자리합니다. 하위 질문을 생성하는 플래너 에이전트, 각 하위 질문에 대해 GPT Researcher의 검색/스크레이프 파이프라인을 호출하는 검색 단계(웹 및 로컬 문서 리트리버를 별도로 구성), 그리고 집계된 결과를 받아 인용이 포함된 보고서 초안을 작성하는 라이터 에이전트를 정의합니다. GPT Researcher의 설정을 통해 보고서 유형, 보고서 길이, 소스 개수를 구성한 다음, 브리프를 AG2 그룹 채팅 또는 순차 파이프라인을 통해 전달합니다. 출력은 마크다운 파일로 작성되며, 선택적으로 제공된 Docker 이미지나 Colab 노트북을 통해 호스팅 경로로 실행할 수 있습니다.
Try it yourself
# Deep Research Agent 설정 가이드
이 저장소에 GPT Researcher와 AG2를 사용하는 딥 리서치 에이전트를 추가합니다. 아래는 파일 구조와 각 파일의 내용입니다.
## 파일 구조
```
.
├── config/
│ └── research.yaml
├── src/
│ └── research_agent/
│ ├── __init__.py
│ ├── config.py
│ ├── agent.py
│ ├── local_docs.py
│ └── cli.py
├── docs/ # 로컬 문서를 여기에 둡니다
├── reports/ # 생성된 리포트 출력 위치
├── pyproject.toml
└── README.md
```
## `config/research.yaml`
```yaml
# 모델 및 검색 설정
model:
provider: openai # openai | anthropic | ollama
name: gpt-4o-mini
temperature: 0.2
search:
provider: tavily # tavily | duckduckgo | serper
max_results_per_query: 8
research:
max_sub_questions: 6
max_iterations: 3
report_type: detailed_report
local_docs:
enabled: true
paths:
- ./docs
extensions: [".md", ".txt", ".pdf"]
chunk_size: 1200
chunk_overlap: 200
```
## `src/research_agent/config.py`
```python
from __future__ import annotations
from pathlib import Path
from typing import Any
import yaml
from pydantic import BaseModel, Field
class ModelConfig(BaseModel):
provider: str = "openai"
name: str = "gpt-4o-mini"
temperature: float = 0.2
class SearchConfig(BaseModel):
provider: str = "tavily"
max_results_per_query: int = 8
class ResearchConfig(BaseModel):
max_sub_questions: int = 6
max_iterations: int = 3
report_type: str = "detailed_report"
class LocalDocsConfig(BaseModel):
enabled: bool = True
paths: list[str] = Field(default_factory=lambda: ["./docs"])
extensions: list[str] = Field(default_factory=lambda: [".md", ".txt", ".pdf"])
chunk_size: int = 1200
chunk_overlap: int = 200
class AppConfig(BaseModel):
model: ModelConfig = ModelConfig()
search: SearchConfig = SearchConfig()
research: ResearchConfig = ResearchConfig()
local_docs: LocalDocsConfig = LocalDocsConfig()
def load_config(path: str | Path = "config/research.yaml") -> AppConfig:
p = Path(path)
if not p.exists():
return AppConfig()
data: dict[str, Any] = yaml.safe_load(p.read_text(encoding="utf-8")) or {}
return AppConfig(**data)
```
## `src/research_agent/local_docs.py`
```python
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from research_agent.config import LocalDocsConfig
@dataclass
class DocChunk:
source: str
text: str
def _read_file(path: Path) -> str:
if path.suffix.lower() == ".pdf":
try:
from pypdf import PdfReader
except ImportError as e:
raise RuntimeError("PDF 지원을 위해 `pip install pypdf`가 필요합니다.") from e
reader = PdfReader(str(path))
return "\n".join(page.extract_text() or "" for page in reader.pages)
return path.read_text(encoding="utf-8", errors="ignore")
def _chunk(text: str, size: int, overlap: int) -> list[str]:
chunks: list[str] = []
start = 0
while start < len(text):
end = min(start + size, len(text))
chunks.append(text[start:end])
if end == len(text):
break
start = end - overlap
return chunks
def load_local_docs(cfg: LocalDocsConfig) -> list[DocChunk]:
if not cfg.enabled:
return []
chunks: list[DocChunk] = []
exts = {e.lower() for e in cfg.extensions}
for raw in cfg.paths:
root = Path(raw)
if not root.exists():
continue
files = [root] if root.is_file() else [p for p in root.rglob("*") if p.is_file()]
for f in files:
if f.suffix.lower() not in exts:
continue
text = _read_file(f)
for c in _chunk(text, cfg.chunk_size, cfg.chunk_overlap):
chunks.append(DocChunk(source=str(f), text=c))
return chunks
def search_local(chunks: list[DocChunk], query: str, top_k: int = 5) -> list[DocChunk]:
"""간단한 키워드 기반 검색. 필요 시 임베딩 기반으로 교체하세요."""
terms = [t.lower() for t in query.split() if len(t) > 2]
scored: list[tuple[int, DocChunk]] = []
for c in chunks:
low = c.text.lower()
score = sum(low.count(t) for t in terms)
if score:
scored.append((score, c))
scored.sort(key=lambda x: x[0], reverse=True)
return [c for _, c in scored[:top_k]]
```
## `src/research_agent/agent.py`
```python
from __future__ import annotations
import json
from dataclasses import dataclass
from pathlib import Path
from autogen import AssistantAgent, UserProxyAgent, config_list_from_json
from research_agent.config import AppConfig
from research_agent.local_docs import DocChunk, load_local_docs, search_local
@dataclass
class ResearchResult:
brief: str
sub_questions: list[str]
report_markdown: str
sources: list[str]
PLANNER_SYSTEM = """You are a research planner. Given a research brief, produce a JSON array of
sub-questions that together cover the brief. Return ONLY valid JSON, no prose."""
SYNTH_SYSTEM = """You are a research analyst. Given a brief, sub-questions, and gathered evidence
(web + local docs), write a long-form markdown report. Requirements:
- Use clear section headings.
- Include inline citations like [1], [2] that map to the provided source list.
- End with a '## Sources' section listing each source with its number.
- Do not invent sources; only cite provided evidence."""
def _llm_config(cfg: AppConfig) -> dict:
provider = cfg.model.provider.lower()
if provider == "openai":
return {
"config_list": [{
"model": cfg.model.name,
"api_type": "openai",
"api_key_env": "OPENAI_API_KEY",
}],
"temperature": cfg.model.temperature,
}
if provider == "anthropic":
return {
"config_list": [{
"model": cfg.model.name,
"api_type": "anthropic",
"api_key_env": "ANTHROPIC_API_KEY",
}],
"temperature": cfg.model.temperature,
}
if provider == "ollama":
return {
"config_list": [{
"model": cfg.model.name,
"base_url": "http://localhost:11434/v1",
"api_key": "ollama",
}],
"temperature": cfg.model.temperature,
}
raise ValueError(f"지원하지 않는 모델 provider: {cfg.model.provider}")
def _plan_sub_questions(brief: str, cfg: AppConfig) -> list[str]:
planner = AssistantAgent(
name="planner",
system_message=PLANNER_SYSTEM,
llm_config=_llm_config(cfg),
)
user = UserProxyAgent(
name="user",
human_input_mode="NEVER",
code_execution_config=False,
max_consecutive_auto_reply=0,
)
prompt = (
f"Research brief:\n{brief}\n\n"
f"Produce at most {cfg.research.max_sub_questions} sub-questions as a JSON array."
)
user.initiate_chat(planner, message=prompt, max_turns=1)
raw = planner.last_message()["content"].strip()
# JSON 블록만 추출
start, end = raw.find("["), raw.rfind("]")
if start == -1 or end == -1:
return [brief]
try:
questions = json.loads(raw[start : end + 1])
return [str(q) for q in questions][: cfg.research.max_sub_questions]
except json.JSONDecodeError:
return [brief]
def _web_search(query: str, cfg: AppConfig) -> list[dict]:
provider = cfg.search.provider.lower()
if provider == "tavily":
from tavily import TavilyClient
client = TavilyClient()
res = client.search(query, max_results=cfg.search.max_results_per_query)
return [
{"title": r.get("title", ""), "url": r.get("url", ""), "content": r.get("content", "")}
for r in res.get("results", [])
]
if provider == "duckduckgo":
from duckduckgo_search import DDGS
with DDGS() as ddgs:
return [
{"title": r.get("title", ""), "url": r.get("href", ""), "content": r.get("body", "")}
for r in ddgs.text(query, max_results=cfg.search.max_results_per_query)
]
if provider == "serper":
import os, requests
resp = requests.post(
"https://google.serper.dev/search",
headers={"X-API-KEY": os.environ["SERPER_API_KEY"]},
json={"q": query, "num": cfg.search.max_results_per_query},
timeout=30,
)
resp.raise_for_status()
return [
{"title": r.get("title", ""), "url": r.get("link", ""), "content": r.get("snippet", "")}
for r in resp.json().get("organic", [])
]
raise ValueError(f"지원하지 않는 검색 provider: {cfg.search.provider}")
def _gather_evidence(
sub_questions: list[str],
local_chunks: list[DocChunk],
cfg: AppConfig,
) -> tuple[list[dict], list[str]]:
evidence: list[dict] = []
sources: list[str] = []
seen_urls: set[str] = set()
for q in sub_questions:
for r in _web_search(q, cfg):
if r["url"] and r["url"] not in seen_urls:
seen_urls.add(r["url"])
evidence.append({"kind": "web", "query": q, **r})
sources.append(f"{r['title']} — {r['url']}")
for c in search_local(local_chunks, q):
evidence.append({"kind": "local", "query": q, "source": c.source, "content": c.text})
label = f"Local: {c.source}"
if label not in sources:
sources.append(label)
return evidence, sources
def _synthesize(
brief: str,
sub_questions: list[str],
evidence: list[dict],
sources: list[str],
cfg: AppConfig,
) -> str:
writer = AssistantAgent(
name="writer",
system_message=SYNTH_SYSTEM,
llm_config=_llm_config(cfg),
)
user = UserProxyAgent(
name="user",
human_input_mode="NEVER",
code_execution_config=False,
max_consecutive_auto_reply=0,
)
numbered_sources = "\n".join(f"[{i+1}] {s}" for i, s in enumerate(sources))
evidence_block = "\n\n".join(
f"[{i+1}] ({e['kind']}) {e.get('title', e.get('source', ''))}\n{e.get('content', '')[:1500]}"
for i, e in enumerate(evidence)
)
prompt = (
f"Research brief:\n{brief}\n\n"
f"Sub-questions:\n" + "\n".join(f"- {q}" for q in sub_questions) + "\n\n"
f"Evidence:\n{evidence_block}\n\n"
f"Source list (use these numbers for citations):\n{numbered_sources}\n\n"
"Write the final markdown report now."
)
user.initiate_chat(writer, message=prompt, max_turns=1)
return writer.last_message()["content"]
def run_research(brief: str, cfg: AppConfig) -> ResearchResult:
local_chunks = load_local_docs(cfg.local_docs)
sub_questions = _plan_sub_questions(brief, cfg)
evidence, sources = _gather_evidence(sub_questions, local_chunks, cfg)
report = _synthesize(brief, sub_questions, evidence, sources, cfg)
return ResearchResult(
brief=brief,
sub_questions=sub_questions,
report_markdown=report,
sources=sources,
)
def write_report(result: ResearchResult, out_dir: str | Path = "./reports") -> Path:
out = Path(out_dir)
out.mkdir(parents=True, exist_ok=True)
slug = "".join(c if c.isalnum() or c in "-_" else "-" for c in result.brief[:60]).strip("-")
path = out / f"{slug or 'report'}.md"
path.write_text(result.report_markdown, encoding="utf-8")
return path
```
## `src/research_agent/cli.py`
```python
from __future__ import annotations
import argparse
import sys
from pathlib import Path
from research_agent.agent import run_research, write_report
from research_agent.config import load_config
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Deep research agent (GPT Researcher + AG2)")
parser.add_argument("brief", help="Research brief (quoted string)")
parser.add_argument("--config", default="config/research.yaml", help="Path to config YAML")
parser.add_argument("--out", default="./reports", help="Output directory for reports")
args = parser.parse_args(argv)
cfg = load_config(args.config)
result = run_research(args.brief, cfg)
path = write_report(result, args.out)
print(f"Report written to {path}")
return 0
if __name__ == "__main__":
sys.exit(main())
```
## `src/research_agent/__init__.py`
```python
from research_agent.agent import run_research, write_report
from research_agent.config import load_config
__all__ = ["run_research", "write_report", "load_config"]
```
## `pyproject.toml`
```toml
[project]
name = "research-agent"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = [
"ag2>=0.5",
"gpt-researcher>=0.8",
"pyyaml>=6.0",
"pydantic>=2.0",
"tavily-python>=0.3",
"duckduckgo-search>=6.0",
"requests>=2.31",
"pypdf>=4.0",
]
[project.scripts]
research = "research_agent.cli:main"
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
[tool.setuptools.packages.find]
where = ["src"]
```
## `README.md` (추가 섹션)
```markdown
## Deep Research Agent
GPT Researcher와 AG2를 사용하는 딥 리서치 에이전트입니다. 리서치 브리프를 받아
하위 질문으로 분해하고, 웹과 로컬 문서를 검색한 뒤 인라인 인용과 소스 목록이 포함된
장문 마크다운 리포트를 생성합니다.
### 설치
```bash
pip install -e .
export OPENAI_API_KEY=... # 또는 ANTHROPIC_API_KEY
export TAVILY_API_KEY=... # search.provider=tavily 인 경우
```
### 사용
```bash
research "2025년 LLM 에이전트 프레임워크 동향과 실무 적용 사례" --config config/research.yaml
```
리포트는 `./reports/<slug>.md`에 저장됩니다.
### 로컬 문서 추가하기
1. 문서를 `./docs` 아래에 두거나, `config/research.yaml`의 `local_docs.paths`에
원하는 디렉터리/파일 경로를 추가합니다.
2. 지원 확장자는 `local_docs.extensions`에서 설정합니다 (기본: `.md`, `.txt`, `.pdf`).
3. 청크 크기는 `chunk_size` / `chunk_overlap`으로 조정합니다.
4. `local_docs.enabled: false`로 설정하면 로컬 검색을 끕니다.
로컬 문서는 하위 질문별 키워드 매칭으로 검색되며, 리포트의 소스 목록에
`Local: <경로>` 형태로 표시됩니다. 임베딩 기반 검색이 필요하면
`src/research_agent/local_docs.py`의 `search_local`을 교체하세요.
```
## 참고 사항
- `gpt-researcher`는 자체 파이프라인을 제공하지만, 여기서는 AG2 에이전트로 계획·수집·합성을 직접 오케스트레이션합니다. 필요 시 `gpt-researcher`의 `GPTResearcher`를 `_gather_evidence` 내부에서 호출해 웹 수집을 위임할 수 있습니다.
- `search_local`은 의존성을 줄이기 위한 키워드 기반 구현입니다. 품질이 중요하면 `sentence-transformers`나 `chromadb`로 교체하세요.
- API 키는 환경 변수로만 주입하고, `config/research.yaml`에는 커밋하지 마세요.