1484 lines
54 KiB
Markdown
1484 lines
54 KiB
Markdown
# Deep-Research → Codex/Antigravity CLI 이식 Implementation Plan
|
||
|
||
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||
|
||
**Goal:** Claude Code의 deep-research 하네스를 codex-cli·antigravity-cli 각각이 그대로 수행하도록 단일 외부 Python 드라이버로 이식한다.
|
||
|
||
**Architecture:** 결정론 제어 로직(dedup·예산·3-vote 정족수·랭킹)을 `core.py` 순수함수로 1:1 이식하고, 플랫폼 차이는 `backends/`(codex/antigravity) 어댑터로 흡수한다. 에이전트 실행기는 각 CLI의 headless 모드(`codex exec` / `agy -p`)이고 웹 조사는 각 CLI 네이티브 도구가 수행한다. MockBackend로 네트워크 0 상태에서 충실도를 단위/통합 테스트한다.
|
||
|
||
**Tech Stack:** Python 3.11+, Pydantic v2, asyncio/subprocess(stdlib), pytest, pytest-asyncio. 외부 검색 API·SDK 없음(CLI 구독으로 충당).
|
||
|
||
**Spec:** `docs/superpowers/specs/2026-06-09-deep-research-codex-antigravity-port-design.md`
|
||
**원본 JS(충실도 기준):** `~/.claude/projects/-home-donghyeon-dev-llm-wiki-private/88afa9ca-1e45-4353-9ff0-6361812d9053/workflows/scripts/deep-research-wf_aecef33f-4cc.js`
|
||
|
||
---
|
||
|
||
## File Structure
|
||
|
||
| 파일 | 책임 |
|
||
|---|---|
|
||
| `scripts/deep-research/pyproject.toml` | 패키지 메타·deps |
|
||
| `scripts/deep-research/deep_research/config.py` | 상수(원본 동일) + 동시성 cap |
|
||
| `scripts/deep-research/deep_research/schemas.py` | 5 Pydantic 모델 + JSON Schema export |
|
||
| `scripts/deep-research/deep_research/prompts.py` | 6단계 프롬프트(원본 문자열 이식) |
|
||
| `scripts/deep-research/deep_research/core.py` | 결정론 순수함수: norm_url/Deduper/rank_claims/tally/stats/degenerate |
|
||
| `scripts/deep-research/deep_research/pipeline.py` | asyncio 오케스트레이션(무배리어 pipeline / verify 배리어) |
|
||
| `scripts/deep-research/deep_research/backends/base.py` | AgentBackend ABC |
|
||
| `scripts/deep-research/deep_research/backends/mock.py` | 테스트 더블 |
|
||
| `scripts/deep-research/deep_research/backends/codex.py` | `codex exec --json --output-schema` |
|
||
| `scripts/deep-research/deep_research/backends/antigravity.py` | `agy -p` + 검증·재시도 |
|
||
| `scripts/deep-research/deep_research/report.py` | Report → markdown/JSON |
|
||
| `scripts/deep-research/deep_research/__main__.py` | CLI 진입점 |
|
||
| `scripts/deep-research/tests/test_core.py` | core 순수함수 단위테스트 |
|
||
| `scripts/deep-research/tests/test_pipeline_mock.py` | MockBackend 통합 + 퇴화 경로 3종 |
|
||
| `scripts/deep-research/README.md` | 사용법·인증 전제 |
|
||
|
||
> **공통 규칙**: LLM이 채우는 스키마 필드는 원본 JSON 키와 동일하게 **camelCase 속성명**(`sourceQuality`, `publishDate`, `counterSource`, `openQuestions`, `sourceUrl`)을 쓴다 — 프롬프트 지시·원본 wire 포맷과 1:1 일치시키기 위함. core 내부 dict도 동일 키 사용.
|
||
|
||
---
|
||
|
||
### Task 0: 프로젝트 스캐폴드
|
||
|
||
**Files:**
|
||
- Create: `scripts/deep-research/pyproject.toml`
|
||
- Create: `scripts/deep-research/deep_research/__init__.py`
|
||
- Create: `scripts/deep-research/deep_research/backends/__init__.py`
|
||
- Create: `scripts/deep-research/tests/__init__.py`
|
||
|
||
- [ ] **Step 1: pyproject.toml 작성**
|
||
|
||
```toml
|
||
[project]
|
||
name = "deep-research-driver"
|
||
version = "0.1.0"
|
||
description = "Port of Claude Code deep-research harness to Codex/Antigravity CLIs"
|
||
requires-python = ">=3.11"
|
||
dependencies = ["pydantic>=2.6"]
|
||
|
||
[project.optional-dependencies]
|
||
dev = ["pytest>=8", "pytest-asyncio>=0.23"]
|
||
|
||
[tool.pytest.ini_options]
|
||
asyncio_mode = "auto"
|
||
testpaths = ["tests"]
|
||
|
||
[build-system]
|
||
requires = ["setuptools>=68"]
|
||
build-backend = "setuptools.build_meta"
|
||
```
|
||
|
||
- [ ] **Step 2: 빈 `__init__.py` 3개 생성** (`deep_research/`, `deep_research/backends/`, `tests/`) — 내용 없음.
|
||
|
||
- [ ] **Step 3: 가상환경·설치 검증**
|
||
|
||
Run: `cd scripts/deep-research && python -m venv .venv && .venv/bin/pip install -e ".[dev]"`
|
||
Expected: 설치 성공, `pytest` 사용 가능.
|
||
|
||
- [ ] **Step 4: Commit**
|
||
|
||
```bash
|
||
git add scripts/deep-research/pyproject.toml scripts/deep-research/deep_research/__init__.py scripts/deep-research/deep_research/backends/__init__.py scripts/deep-research/tests/__init__.py
|
||
git commit -m "chore(deep-research): scaffold python driver package"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 1: config.py (상수)
|
||
|
||
**Files:**
|
||
- Create: `scripts/deep-research/deep_research/config.py`
|
||
|
||
- [ ] **Step 1: config 작성 (원본 상수 1:1)**
|
||
|
||
```python
|
||
"""원본 deep-research-wf JS 와 동일한 상수. 변경 금지(충실도)."""
|
||
from dataclasses import dataclass
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Config:
|
||
VOTES_PER_CLAIM: int = 3
|
||
REFUTATIONS_REQUIRED: int = 2
|
||
MAX_FETCH: int = 15
|
||
MAX_VERIFY_CLAIMS: int = 25
|
||
CONCURRENCY: int = 10 # asyncio.Semaphore (원본 Workflow cap 대응)
|
||
ANTIGRAVITY_RETRIES: int = 2 # 스키마 검증 실패 시 재프롬프트 횟수
|
||
|
||
|
||
DEFAULT = Config()
|
||
```
|
||
|
||
- [ ] **Step 2: import 검증**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/python -c "from deep_research.config import DEFAULT; print(DEFAULT.MAX_FETCH)"`
|
||
Expected: `15`
|
||
|
||
- [ ] **Step 3: Commit**
|
||
|
||
```bash
|
||
git add scripts/deep-research/deep_research/config.py
|
||
git commit -m "feat(deep-research): add config constants (1:1 with JS harness)"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 2: schemas.py (5 Pydantic 모델)
|
||
|
||
**Files:**
|
||
- Create: `scripts/deep-research/deep_research/schemas.py`
|
||
- Test: `scripts/deep-research/tests/test_core.py` (이 태스크에서 스키마 검증 테스트만 먼저)
|
||
|
||
- [ ] **Step 1: 스키마 검증 실패 테스트 작성**
|
||
|
||
`tests/test_core.py` 에 추가:
|
||
|
||
```python
|
||
import pytest
|
||
from pydantic import ValidationError
|
||
from deep_research.schemas import Scope, Angle, Search, SearchResult, Extract, Verdict, Report
|
||
|
||
|
||
def test_scope_requires_min_3_angles():
|
||
with pytest.raises(ValidationError):
|
||
Scope(question="q", summary="s", angles=[Angle(label="a", query="x")])
|
||
|
||
|
||
def test_scope_accepts_5_angles():
|
||
angles = [Angle(label=f"a{i}", query="x") for i in range(5)]
|
||
s = Scope(question="q", summary="s", angles=angles)
|
||
assert len(s.angles) == 5
|
||
|
||
|
||
def test_search_relevance_enum_enforced():
|
||
with pytest.raises(ValidationError):
|
||
SearchResult(url="http://a", title="t", relevance="bogus")
|
||
|
||
|
||
def test_json_schema_export_works():
|
||
schema = Scope.model_json_schema()
|
||
assert "angles" in schema["properties"]
|
||
```
|
||
|
||
- [ ] **Step 2: 테스트 실패 확인**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/pytest tests/test_core.py -k "scope or relevance or json_schema" -v`
|
||
Expected: FAIL — `ModuleNotFoundError: deep_research.schemas`
|
||
|
||
- [ ] **Step 3: schemas.py 구현**
|
||
|
||
```python
|
||
"""원본 5 SCHEMA 의 Pydantic v2 포팅. LLM-facing 필드는 원본 JSON 키(camelCase) 유지."""
|
||
from typing import Literal, Optional
|
||
from pydantic import BaseModel, Field
|
||
|
||
|
||
class Angle(BaseModel):
|
||
label: str
|
||
query: str
|
||
rationale: Optional[str] = None
|
||
|
||
|
||
class Scope(BaseModel):
|
||
question: str
|
||
summary: str
|
||
angles: list[Angle] = Field(min_length=3, max_length=6)
|
||
|
||
|
||
class SearchResult(BaseModel):
|
||
url: str
|
||
title: str
|
||
snippet: Optional[str] = None
|
||
relevance: Literal["high", "medium", "low"]
|
||
|
||
|
||
class Search(BaseModel):
|
||
results: list[SearchResult] = Field(max_length=6)
|
||
|
||
|
||
class Claim(BaseModel):
|
||
claim: str
|
||
quote: str
|
||
importance: Literal["central", "supporting", "tangential"]
|
||
|
||
|
||
class Extract(BaseModel):
|
||
sourceQuality: Literal["primary", "secondary", "blog", "forum", "unreliable"]
|
||
publishDate: Optional[str] = None
|
||
claims: list[Claim] = Field(max_length=5)
|
||
|
||
|
||
class Verdict(BaseModel):
|
||
refuted: bool
|
||
evidence: str
|
||
confidence: Literal["high", "medium", "low"]
|
||
counterSource: Optional[str] = None
|
||
|
||
|
||
class Finding(BaseModel):
|
||
claim: str
|
||
confidence: Literal["high", "medium", "low"]
|
||
sources: list[str]
|
||
evidence: str
|
||
vote: Optional[str] = None
|
||
|
||
|
||
class Report(BaseModel):
|
||
summary: str
|
||
findings: list[Finding]
|
||
caveats: str
|
||
openQuestions: Optional[list[str]] = None
|
||
```
|
||
|
||
- [ ] **Step 4: 테스트 통과 확인**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/pytest tests/test_core.py -k "scope or relevance or json_schema" -v`
|
||
Expected: PASS (4 passed)
|
||
|
||
- [ ] **Step 5: Commit**
|
||
|
||
```bash
|
||
git add scripts/deep-research/deep_research/schemas.py scripts/deep-research/tests/test_core.py
|
||
git commit -m "feat(deep-research): add 5 Pydantic schemas (1:1 with JS SCHEMA)"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 3: prompts.py (6단계 프롬프트 이식)
|
||
|
||
**Files:**
|
||
- Create: `scripts/deep-research/deep_research/prompts.py`
|
||
|
||
> 원본 JS 의 프롬프트 문자열을 그대로 이식한다. 함수는 인자를 받아 문자열을 만든다.
|
||
|
||
- [ ] **Step 1: prompts.py 구현 (원본 문자열 1:1)**
|
||
|
||
```python
|
||
"""원본 deep-research-wf JS 의 프롬프트 문자열 이식. 표현 변경 금지(충실도)."""
|
||
|
||
|
||
def scope_prompt(question: str) -> str:
|
||
return (
|
||
"Decompose this research question into complementary search angles.\n\n"
|
||
"## Question\n" + question + "\n\n"
|
||
"## Task\n"
|
||
"Generate 5 distinct web search queries that together cover the question from "
|
||
"different angles. Pick angles that suit the question's domain. Examples:\n"
|
||
"- broad/primary · academic/technical · recent news · contrarian/skeptical · practitioner/implementation\n"
|
||
"- For medical: anatomy · common causes · serious differentials · authoritative refs · red flags\n"
|
||
"- For tech: state-of-art · benchmarks · limitations · industry adoption · cost/tradeoffs\n\n"
|
||
"Make queries specific enough to surface high-signal results. Avoid redundancy.\n"
|
||
"Return: the question (verbatim or lightly normalized), a 1-2 sentence decomposition "
|
||
"strategy, and the angles.\n\nStructured output only."
|
||
)
|
||
|
||
|
||
def search_prompt(question: str, angle) -> str:
|
||
return (
|
||
"## Web Searcher: " + angle.label + "\n\n"
|
||
'Research question: "' + question + '"\n\n'
|
||
"Your angle: **" + angle.label + "** — " + (angle.rationale or "") + "\n"
|
||
"Search query: `" + angle.query + "`\n\n"
|
||
"## Task\nUse web search with the query above (or a refined version). Return the top "
|
||
"4-6 most relevant results.\n"
|
||
"Rank by relevance to the ORIGINAL question, not just the search query. Skip obvious "
|
||
"SEO spam/content farms.\n"
|
||
"Include a short snippet capturing why each result is relevant.\n\nStructured output only."
|
||
)
|
||
|
||
|
||
def fetch_prompt(question: str, source: dict, angle: str) -> str:
|
||
return (
|
||
"## Source Extractor\n\n"
|
||
'Research question: "' + question + '"\n\n'
|
||
"Fetch and extract key claims from this source:\n"
|
||
"**URL:** " + source["url"] + "\n**Title:** " + source["title"] + "\n**Found via:** " + angle + " search\n\n"
|
||
"## Task\n1. Fetch the page content.\n"
|
||
"2. Assess source quality: primary research/institution? secondary reporting? blog/opinion? forum? unreliable?\n"
|
||
"3. Extract 2-5 FALSIFIABLE claims that bear on the research question. Each claim must:\n"
|
||
" - be a concrete, checkable statement (not vague generalities)\n"
|
||
" - include a direct quote from the source as support\n"
|
||
" - be rated central/supporting/tangential to the research question\n"
|
||
"4. Note publish date if available.\n\n"
|
||
'If the fetch fails or the page is irrelevant/paywalled, return claims: [] and '
|
||
'sourceQuality: "unreliable".\n\nStructured output only.'
|
||
)
|
||
|
||
|
||
def verify_prompt(question: str, claim: dict, v: int, votes_per_claim: int, refutations_required: int) -> str:
|
||
return (
|
||
"## Adversarial Claim Verifier (voter " + str(v + 1) + "/" + str(votes_per_claim) + ")\n\n"
|
||
"Be SKEPTICAL. Try to REFUTE this claim. ≥" + str(refutations_required) + "/" + str(votes_per_claim) + " refutations kill it.\n\n"
|
||
"## Research question\n" + question + "\n\n"
|
||
'## Claim under review\n"' + claim["claim"] + '"\n\n'
|
||
"**Source:** " + claim["sourceUrl"] + " (" + claim["sourceQuality"] + ")\n"
|
||
'**Supporting quote:** "' + claim["quote"] + '"\n\n'
|
||
"## Checklist\n"
|
||
"1. Is the claim actually supported by the quote, or is it an overreach/misread?\n"
|
||
"2. Search for contradicting evidence — does any credible source dispute or heavily qualify this?\n"
|
||
"3. Is the source quality sufficient for the claim's strength? (extraordinary claims need primary sources)\n"
|
||
"4. Is the claim outdated? (check dates — old claims about fast-moving fields are suspect)\n"
|
||
"5. Is this a marketing claim / press release / cherry-picked benchmark / forum speculation?\n\n"
|
||
"**refuted=true** if: unsupported by quote / contradicted / low-quality source for strong claim / outdated / marketing fluff.\n"
|
||
"**refuted=false** ONLY if: claim is well-supported, current, and source quality matches claim strength.\n"
|
||
"Default to refuted=true if uncertain.\n\nStructured output only. Evidence MUST be specific."
|
||
)
|
||
|
||
|
||
def synth_prompt(question: str, block: str, killed_block: str, confirmed_count: int, votes_per_claim: int) -> str:
|
||
return (
|
||
"## Synthesis: research report\n\n"
|
||
"**Question:** " + question + "\n\n"
|
||
+ str(confirmed_count) + " claims survived " + str(votes_per_claim) + "-vote adversarial verification. "
|
||
"Merge semantic duplicates and synthesize.\n\n"
|
||
"## Confirmed claims\n" + block + "\n" + killed_block + "\n\n"
|
||
"## Instructions\n"
|
||
"1. Identify claims that say the same thing — merge them, combine their sources.\n"
|
||
"2. Group related claims into coherent findings. Each finding should directly address the research question.\n"
|
||
"3. Assign confidence per finding: high (multiple primary sources, unanimous votes), medium (secondary sources or split votes), low (single source or blog-quality).\n"
|
||
"4. Write a 3-5 sentence executive summary answering the research question.\n"
|
||
"5. Note caveats: what's uncertain, what sources were weak, what time-sensitivity applies.\n"
|
||
"6. List 2-4 open questions that emerged but weren't answered.\n\nStructured output only."
|
||
)
|
||
```
|
||
|
||
- [ ] **Step 2: import 검증**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/python -c "from deep_research import prompts; print(prompts.scope_prompt('q')[:20])"`
|
||
Expected: `Decompose this resea`
|
||
|
||
- [ ] **Step 3: Commit**
|
||
|
||
```bash
|
||
git add scripts/deep-research/deep_research/prompts.py
|
||
git commit -m "feat(deep-research): port 6-phase prompt strings from JS harness"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 4: core — norm_url + Deduper (TDD)
|
||
|
||
**Files:**
|
||
- Create: `scripts/deep-research/deep_research/core.py`
|
||
- Test: `scripts/deep-research/tests/test_core.py` (추가)
|
||
|
||
- [ ] **Step 1: 실패 테스트 작성** — `tests/test_core.py` 에 추가:
|
||
|
||
```python
|
||
from deep_research.core import norm_url, Deduper
|
||
from deep_research.config import Config
|
||
|
||
|
||
def test_norm_url_strips_www_and_trailing_slash():
|
||
assert norm_url("http://www.Example.com/Path/") == "example.com/path"
|
||
|
||
|
||
def test_norm_url_bare_host():
|
||
assert norm_url("https://example.com") == "example.com"
|
||
|
||
|
||
def test_norm_url_fallback_on_garbage():
|
||
assert norm_url("not a url") == "not a url"
|
||
|
||
|
||
def test_deduper_filters_exact_dupes():
|
||
d = Deduper(Config(MAX_FETCH=15))
|
||
r = [{"url": "http://a.com/x", "title": "t", "relevance": "high"}]
|
||
assert len(d.filter_novel("angle1", r)) == 1
|
||
# same normalized url from another angle -> dup
|
||
r2 = [{"url": "http://www.a.com/x/", "title": "t2", "relevance": "high"}]
|
||
assert d.filter_novel("angle2", r2) == []
|
||
assert len(d.dupes) == 1
|
||
|
||
|
||
def test_deduper_budget_drops_medium_low_when_slots_exhausted():
|
||
d = Deduper(Config(MAX_FETCH=1))
|
||
first = [{"url": "http://a.com/1", "title": "t", "relevance": "high"}]
|
||
d.filter_novel("a1", first) # consumes the only slot
|
||
more = [
|
||
{"url": "http://b.com/2", "title": "t", "relevance": "medium"},
|
||
{"url": "http://c.com/3", "title": "t", "relevance": "low"},
|
||
]
|
||
assert d.filter_novel("a2", more) == []
|
||
assert len(d.budget_dropped) == 2
|
||
|
||
|
||
def test_deduper_high_passes_even_when_slots_exhausted():
|
||
# 원본: high(rank 0)는 budget 조건(rank>=1)에 안 걸려 slot<=0 이어도 통과
|
||
d = Deduper(Config(MAX_FETCH=1))
|
||
d.filter_novel("a1", [{"url": "http://a.com/1", "title": "t", "relevance": "high"}])
|
||
high = [{"url": "http://d.com/4", "title": "t", "relevance": "high"}]
|
||
assert len(d.filter_novel("a2", high)) == 1
|
||
assert d.fetch_slots == -1
|
||
```
|
||
|
||
- [ ] **Step 2: 테스트 실패 확인**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/pytest tests/test_core.py -k "norm_url or deduper" -v`
|
||
Expected: FAIL — `ImportError: cannot import name 'norm_url'`
|
||
|
||
- [ ] **Step 3: core.py 구현 (1차)**
|
||
|
||
```python
|
||
"""결정론 제어 로직 — 원본 JS 와 1:1. 네트워크·LLM 의존 0."""
|
||
from urllib.parse import urlparse
|
||
from deep_research.config import Config
|
||
|
||
REL_RANK = {"high": 0, "medium": 1, "low": 2}
|
||
IMP_RANK = {"central": 0, "supporting": 1, "tangential": 2}
|
||
QUAL_RANK = {"primary": 0, "secondary": 1, "blog": 2, "forum": 3, "unreliable": 4}
|
||
|
||
|
||
def norm_url(u: str) -> str:
|
||
try:
|
||
p = urlparse(u)
|
||
host = (p.hostname or "")
|
||
if not host:
|
||
return u.lower()
|
||
if host.startswith("www."):
|
||
host = host[4:]
|
||
path = (p.path or "").rstrip("/")
|
||
return (host + path).lower()
|
||
except Exception:
|
||
return u.lower()
|
||
|
||
|
||
def host_of(u: str) -> str:
|
||
try:
|
||
h = urlparse(u).hostname or "unknown"
|
||
return h[4:] if h.startswith("www.") else h
|
||
except Exception:
|
||
return "unknown"
|
||
|
||
|
||
class Deduper:
|
||
"""원본 pipeline stage-2 의 dedup+budget 로직. 호출 순서에 결정론적."""
|
||
|
||
def __init__(self, config: Config):
|
||
self.seen: dict[str, dict] = {}
|
||
self.dupes: list[dict] = []
|
||
self.budget_dropped: list[dict] = []
|
||
self.fetch_slots = config.MAX_FETCH
|
||
|
||
def filter_novel(self, angle: str, results: list[dict]) -> list[dict]:
|
||
ordered = sorted(results, key=lambda r: REL_RANK[r["relevance"]])
|
||
novel: list[dict] = []
|
||
for r in ordered:
|
||
key = norm_url(r["url"])
|
||
if key in self.seen:
|
||
self.dupes.append({**r, "angle": angle, "dupOf": self.seen[key]})
|
||
continue
|
||
if self.fetch_slots <= 0 and REL_RANK[r["relevance"]] >= 1:
|
||
self.budget_dropped.append({**r, "angle": angle})
|
||
continue
|
||
self.seen[key] = {"angle": angle, "title": r["title"]}
|
||
self.fetch_slots -= 1
|
||
novel.append(r)
|
||
return novel
|
||
```
|
||
|
||
- [ ] **Step 4: 테스트 통과 확인**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/pytest tests/test_core.py -k "norm_url or deduper" -v`
|
||
Expected: PASS (6 passed)
|
||
|
||
- [ ] **Step 5: Commit**
|
||
|
||
```bash
|
||
git add scripts/deep-research/deep_research/core.py scripts/deep-research/tests/test_core.py
|
||
git commit -m "feat(deep-research): core norm_url + Deduper (TDD, 1:1 with JS)"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 5: core — rank_claims + tally (TDD, 정족수 엣지 고정)
|
||
|
||
**Files:**
|
||
- Modify: `scripts/deep-research/deep_research/core.py`
|
||
- Test: `scripts/deep-research/tests/test_core.py` (추가)
|
||
|
||
- [ ] **Step 1: 실패 테스트 작성** — `tests/test_core.py` 에 추가:
|
||
|
||
```python
|
||
from deep_research.core import rank_claims, tally
|
||
|
||
|
||
def _claim(imp, qual, name="c"):
|
||
return {"claim": name, "importance": imp, "sourceQuality": qual}
|
||
|
||
|
||
def test_rank_claims_orders_by_importance_then_quality():
|
||
claims = [
|
||
_claim("tangential", "primary", "t-prim"),
|
||
_claim("central", "blog", "c-blog"),
|
||
_claim("central", "primary", "c-prim"),
|
||
]
|
||
ranked = rank_claims(claims, max_verify=25)
|
||
assert [c["claim"] for c in ranked] == ["c-prim", "c-blog", "t-prim"]
|
||
|
||
|
||
def test_rank_claims_truncates_to_max():
|
||
claims = [_claim("central", "primary", f"c{i}") for i in range(30)]
|
||
assert len(rank_claims(claims, max_verify=25)) == 25
|
||
|
||
|
||
def test_tally_survives_2_valid_0_refute():
|
||
v = [{"refuted": False}, {"refuted": False}, {"refuted": False}]
|
||
t = tally(v, votes_per_claim=3, refutations_required=2)
|
||
assert t["survives"] is True and t["refutedVotes"] == 0
|
||
|
||
|
||
def test_tally_killed_2_refute():
|
||
v = [{"refuted": True}, {"refuted": True}, {"refuted": False}]
|
||
t = tally(v, votes_per_claim=3, refutations_required=2)
|
||
assert t["survives"] is False and t["refutedVotes"] == 2
|
||
|
||
|
||
def test_tally_all_abstain_does_not_survive():
|
||
# ⚠️ 거짓 생존 차단: all-None -> refuted=0 이지만 valid<2 라 미생존
|
||
t = tally([None, None, None], votes_per_claim=3, refutations_required=2)
|
||
assert t["survives"] is False and t["abstained"] == 3
|
||
|
||
|
||
def test_tally_one_valid_two_abstain_does_not_survive():
|
||
t = tally([{"refuted": False}, None, None], votes_per_claim=3, refutations_required=2)
|
||
assert t["survives"] is False # valid(1) < 2
|
||
```
|
||
|
||
- [ ] **Step 2: 테스트 실패 확인**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/pytest tests/test_core.py -k "rank_claims or tally" -v`
|
||
Expected: FAIL — `ImportError: cannot import name 'rank_claims'`
|
||
|
||
- [ ] **Step 3: core.py 에 추가**
|
||
|
||
```python
|
||
def rank_claims(claims: list[dict], max_verify: int) -> list[dict]:
|
||
return sorted(
|
||
claims,
|
||
key=lambda c: (IMP_RANK[c["importance"]], QUAL_RANK[c["sourceQuality"]]),
|
||
)[:max_verify]
|
||
|
||
|
||
def tally(verdicts: list, votes_per_claim: int, refutations_required: int) -> dict:
|
||
"""원본 verify 정족수 산식 1:1. None=기권."""
|
||
valid = [v for v in verdicts if v is not None]
|
||
refuted = sum(1 for v in valid if v["refuted"])
|
||
abstained = votes_per_claim - len(valid)
|
||
survives = len(valid) >= refutations_required and refuted < refutations_required
|
||
return {
|
||
"valid": valid,
|
||
"refutedVotes": refuted,
|
||
"abstained": abstained,
|
||
"survives": survives,
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 4: 테스트 통과 확인**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/pytest tests/test_core.py -k "rank_claims or tally" -v`
|
||
Expected: PASS (6 passed)
|
||
|
||
- [ ] **Step 5: Commit**
|
||
|
||
```bash
|
||
git add scripts/deep-research/deep_research/core.py scripts/deep-research/tests/test_core.py
|
||
git commit -m "feat(deep-research): core rank_claims + tally with abstention edge (TDD)"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 6: core — synthesis 블록 빌더 + stats + 퇴화 경로 빌더 (TDD)
|
||
|
||
**Files:**
|
||
- Modify: `scripts/deep-research/deep_research/core.py`
|
||
- Test: `scripts/deep-research/tests/test_core.py` (추가)
|
||
|
||
> 원본의 synthesis `block`/`killedBlock` 문자열 조립과 최종 stats·퇴화 반환을 순수 함수로 분리.
|
||
|
||
- [ ] **Step 1: 실패 테스트 작성** — `tests/test_core.py` 에 추가:
|
||
|
||
```python
|
||
from deep_research.core import build_synth_blocks, build_stats
|
||
|
||
|
||
def test_build_synth_blocks_includes_vote_and_source():
|
||
confirmed = [{
|
||
"claim": "X causes Y", "quote": "q", "sourceUrl": "http://a", "sourceQuality": "primary",
|
||
"valid": [{"refuted": False, "confidence": "high", "evidence": "e"}],
|
||
"refutedVotes": 0,
|
||
}]
|
||
killed = [{
|
||
"claim": "Z", "sourceUrl": "http://b", "valid": [{"refuted": True}], "refutedVotes": 1,
|
||
}]
|
||
block, killed_block = build_synth_blocks(confirmed, killed)
|
||
assert "X causes Y" in block and "1-0" in block
|
||
assert "Refuted claims" in killed_block and "Z" in killed_block
|
||
|
||
|
||
def test_build_synth_blocks_no_killed():
|
||
block, killed_block = build_synth_blocks([{
|
||
"claim": "X", "quote": "q", "sourceUrl": "http://a", "sourceQuality": "primary",
|
||
"valid": [{"refuted": False, "confidence": "high", "evidence": "e"}], "refutedVotes": 0,
|
||
}], [])
|
||
assert killed_block == ""
|
||
|
||
|
||
def test_build_stats_agent_calls_formula():
|
||
s = build_stats(angles=5, sources=12, claims=20, voted=18, confirmed=10, killed=8,
|
||
after_synth=6, dupes=3, budget_dropped=2, votes_per_claim=3)
|
||
# 1 + angles + sources + voted*votes_per_claim + 1
|
||
assert s["agentCalls"] == 1 + 5 + 12 + 18 * 3 + 1
|
||
assert s["confirmed"] == 10 and s["afterSynthesis"] == 6
|
||
```
|
||
|
||
- [ ] **Step 2: 테스트 실패 확인**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/pytest tests/test_core.py -k "synth_blocks or build_stats" -v`
|
||
Expected: FAIL — `ImportError`
|
||
|
||
- [ ] **Step 3: core.py 에 추가**
|
||
|
||
```python
|
||
CONF_RANK = {"high": 0, "medium": 1, "low": 2}
|
||
|
||
|
||
def _vote_str(c: dict) -> str:
|
||
return str(len(c["valid"]) - c["refutedVotes"]) + "-" + str(c["refutedVotes"])
|
||
|
||
|
||
def build_synth_blocks(confirmed: list[dict], killed: list[dict]) -> tuple[str, str]:
|
||
"""원본 synthesize 의 block / killedBlock 문자열 조립."""
|
||
parts = []
|
||
for i, c in enumerate(confirmed):
|
||
non_refuted = [v for v in c["valid"] if not v["refuted"]]
|
||
best = sorted(non_refuted, key=lambda v: CONF_RANK[v["confidence"]])[0]
|
||
parts.append(
|
||
"### [" + str(i) + "] " + c["claim"] + "\n"
|
||
+ "Vote: " + _vote_str(c) + " · Source: " + c["sourceUrl"] + " (" + c["sourceQuality"] + ")\n"
|
||
+ 'Quote: "' + c["quote"] + '"\nVerifier evidence (' + best["confidence"] + "): " + best["evidence"] + "\n"
|
||
)
|
||
block = "\n".join(parts)
|
||
|
||
if killed:
|
||
killed_block = "\n## Refuted claims (for transparency)\n" + "\n".join(
|
||
'- "' + c["claim"] + '" (' + c["sourceUrl"] + ", vote " + _vote_str(c) + ")"
|
||
for c in killed
|
||
)
|
||
else:
|
||
killed_block = ""
|
||
return block, killed_block
|
||
|
||
|
||
def build_stats(*, angles, sources, claims, voted, confirmed, killed,
|
||
after_synth, dupes, budget_dropped, votes_per_claim) -> dict:
|
||
return {
|
||
"angles": angles,
|
||
"sourcesFetched": sources,
|
||
"claimsExtracted": claims,
|
||
"claimsVerified": voted,
|
||
"confirmed": confirmed,
|
||
"killed": killed,
|
||
"afterSynthesis": after_synth,
|
||
"urlDupes": dupes,
|
||
"budgetDropped": budget_dropped,
|
||
"agentCalls": 1 + angles + sources + (voted * votes_per_claim) + 1,
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 4: 테스트 통과 확인**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/pytest tests/test_core.py -k "synth_blocks or build_stats" -v`
|
||
Expected: PASS (3 passed)
|
||
|
||
- [ ] **Step 5: 전체 core 테스트 회귀 확인**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/pytest tests/test_core.py -v`
|
||
Expected: PASS (이전 태스크 포함 전부)
|
||
|
||
- [ ] **Step 6: Commit**
|
||
|
||
```bash
|
||
git add scripts/deep-research/deep_research/core.py scripts/deep-research/tests/test_core.py
|
||
git commit -m "feat(deep-research): core synth-block builder + stats (TDD)"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 7: backends/base.py + mock.py
|
||
|
||
**Files:**
|
||
- Create: `scripts/deep-research/deep_research/backends/base.py`
|
||
- Create: `scripts/deep-research/deep_research/backends/mock.py`
|
||
|
||
- [ ] **Step 1: base.py 구현**
|
||
|
||
```python
|
||
from abc import ABC, abstractmethod
|
||
from pydantic import BaseModel
|
||
|
||
|
||
class AgentBackend(ABC):
|
||
"""run_agent: 프롬프트를 1개 CLI headless 에이전트로 실행, schema 로 검증된 객체 반환.
|
||
실패/사용자-skip -> None (원본 .filter(Boolean) 의미)."""
|
||
|
||
@abstractmethod
|
||
async def run_agent(self, prompt: str, schema: type[BaseModel], *, label: str) -> BaseModel | None:
|
||
...
|
||
```
|
||
|
||
- [ ] **Step 2: mock.py 구현**
|
||
|
||
```python
|
||
from pydantic import BaseModel
|
||
from deep_research.backends.base import AgentBackend
|
||
|
||
|
||
class MockBackend(AgentBackend):
|
||
"""테스트 더블. label prefix -> 응답객체(또는 None) 매핑. 네트워크·subprocess 0.
|
||
|
||
responses 의 키는 label prefix. 가장 먼저 매칭되는 prefix 의 값을 반환.
|
||
값이 콜러블이면 (prompt, label) 로 호출해 동적 응답 가능."""
|
||
|
||
def __init__(self, responses: dict, default=None):
|
||
self.responses = responses
|
||
self.default = default
|
||
self.calls: list[str] = []
|
||
|
||
async def run_agent(self, prompt: str, schema: type[BaseModel], *, label: str):
|
||
self.calls.append(label)
|
||
for prefix, resp in self.responses.items():
|
||
if label.startswith(prefix):
|
||
return resp(prompt, label) if callable(resp) else resp
|
||
return self.default
|
||
```
|
||
|
||
- [ ] **Step 3: import 검증**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/python -c "from deep_research.backends.mock import MockBackend; print('ok')"`
|
||
Expected: `ok`
|
||
|
||
- [ ] **Step 4: Commit**
|
||
|
||
```bash
|
||
git add scripts/deep-research/deep_research/backends/base.py scripts/deep-research/deep_research/backends/mock.py
|
||
git commit -m "feat(deep-research): AgentBackend ABC + MockBackend test double"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 8: pipeline.py (asyncio 오케스트레이션, MockBackend TDD)
|
||
|
||
**Files:**
|
||
- Create: `scripts/deep-research/deep_research/pipeline.py`
|
||
- Test: `scripts/deep-research/tests/test_pipeline_mock.py`
|
||
|
||
- [ ] **Step 1: happy-path + 퇴화 경로 실패 테스트 작성** — `tests/test_pipeline_mock.py`:
|
||
|
||
```python
|
||
import pytest
|
||
from deep_research.pipeline import run_research
|
||
from deep_research.backends.mock import MockBackend
|
||
from deep_research.config import Config
|
||
from deep_research.schemas import Scope, Angle, Search, SearchResult, Extract, Claim, Verdict, Report
|
||
|
||
CFG = Config(MAX_FETCH=15, MAX_VERIFY_CLAIMS=25)
|
||
|
||
|
||
def _scope():
|
||
return Scope(question="Q", summary="s",
|
||
angles=[Angle(label=f"a{i}", query=f"q{i}") for i in range(3)])
|
||
|
||
|
||
def _search(prompt, label):
|
||
# 각 각도마다 고유 URL 1개
|
||
n = label.split(":")[1]
|
||
return Search(results=[SearchResult(url=f"http://{n}.com/x", title="t", relevance="high")])
|
||
|
||
|
||
def _extract(prompt, label):
|
||
return Extract(sourceQuality="primary",
|
||
claims=[Claim(claim="C-" + label, quote="q", importance="central")])
|
||
|
||
|
||
def _verdict_pass(prompt, label):
|
||
return Verdict(refuted=False, evidence="e", confidence="high")
|
||
|
||
|
||
def _report():
|
||
return Report(summary="done", findings=[], caveats="none")
|
||
|
||
|
||
async def test_happy_path_returns_report_and_stats():
|
||
backend = MockBackend({
|
||
"scope": _scope(),
|
||
"search:": _search,
|
||
"fetch:": _extract,
|
||
"v": _verdict_pass,
|
||
"synthesize": _report(),
|
||
})
|
||
out = await run_research("Q", backend, config=CFG)
|
||
assert out["summary"] == "done"
|
||
assert out["stats"]["confirmed"] == 3 # 3 각도 × 1 claim, 전부 생존
|
||
assert out["stats"]["angles"] == 3
|
||
|
||
|
||
async def test_empty_question_returns_error():
|
||
out = await run_research(" ", MockBackend({}), config=CFG)
|
||
assert "error" in out
|
||
|
||
|
||
async def test_no_claims_degenerate():
|
||
backend = MockBackend({
|
||
"scope": _scope(),
|
||
"search:": _search,
|
||
"fetch:": lambda p, l: Extract(sourceQuality="unreliable", claims=[]),
|
||
})
|
||
out = await run_research("Q", backend, config=CFG)
|
||
assert out["findings"] == [] and out["stats"]["claims"] == 0
|
||
|
||
|
||
async def test_all_refuted_degenerate():
|
||
backend = MockBackend({
|
||
"scope": _scope(),
|
||
"search:": _search,
|
||
"fetch:": _extract,
|
||
"v": lambda p, l: Verdict(refuted=True, evidence="e", confidence="high"),
|
||
})
|
||
out = await run_research("Q", backend, config=CFG)
|
||
assert out["findings"] == [] and out["stats"]["confirmed"] == 0
|
||
assert len(out["refuted"]) == 3
|
||
|
||
|
||
async def test_synth_failure_salvages_confirmed():
|
||
backend = MockBackend({
|
||
"scope": _scope(),
|
||
"search:": _search,
|
||
"fetch:": _extract,
|
||
"v": _verdict_pass,
|
||
"synthesize": None, # 합성 실패
|
||
})
|
||
out = await run_research("Q", backend, config=CFG)
|
||
assert out["findings"] == [] and len(out["confirmed"]) == 3
|
||
```
|
||
|
||
- [ ] **Step 2: 테스트 실패 확인**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/pytest tests/test_pipeline_mock.py -v`
|
||
Expected: FAIL — `ModuleNotFoundError: deep_research.pipeline`
|
||
|
||
- [ ] **Step 3: pipeline.py 구현**
|
||
|
||
```python
|
||
"""asyncio 오케스트레이션. 원본 pipeline(무배리어)/parallel(배리어) 의미 재현."""
|
||
import asyncio
|
||
from deep_research.config import Config
|
||
from deep_research.core import Deduper, host_of, rank_claims, tally, build_synth_blocks, build_stats
|
||
from deep_research.schemas import Scope, Search, Extract, Verdict, Report
|
||
from deep_research import prompts
|
||
|
||
|
||
async def run_research(question: str, backend, *, config: Config) -> dict:
|
||
q = (question or "").strip()
|
||
if not q:
|
||
return {"error": "No research question provided."}
|
||
|
||
scope = await backend.run_agent(prompts.scope_prompt(q), Scope, label="scope")
|
||
if scope is None:
|
||
return {"error": "Scope agent returned no result — cannot decompose the question."}
|
||
|
||
deduper = Deduper(config)
|
||
sem = asyncio.Semaphore(config.CONCURRENCY)
|
||
dedup_lock = asyncio.Lock()
|
||
|
||
async def search_and_fetch(angle) -> list[dict]:
|
||
async with sem:
|
||
sr = await backend.run_agent(
|
||
prompts.search_prompt(q, angle), Search, label="search:" + angle.label)
|
||
if sr is None:
|
||
return []
|
||
async with dedup_lock: # 공유 상태(seen/fetch_slots) 임계구역 직렬화
|
||
novel = deduper.filter_novel(angle.label, [r.model_dump() for r in sr.results])
|
||
|
||
async def fetch_one(source: dict):
|
||
async with sem:
|
||
ext = await backend.run_agent(
|
||
prompts.fetch_prompt(q, source, angle.label),
|
||
Extract, label="fetch:" + host_of(source["url"]))
|
||
if ext is None:
|
||
return None
|
||
return {
|
||
"url": source["url"], "title": source["title"], "angle": angle.label,
|
||
"sourceQuality": ext.sourceQuality, "publishDate": ext.publishDate,
|
||
"claims": [{**c.model_dump(), "sourceUrl": source["url"], "sourceQuality": ext.sourceQuality}
|
||
for c in ext.claims],
|
||
}
|
||
|
||
fetched = await asyncio.gather(*[fetch_one(s) for s in novel])
|
||
return [f for f in fetched if f is not None]
|
||
|
||
per_angle = await asyncio.gather(*[search_and_fetch(a) for a in scope.angles])
|
||
all_sources = [s for sub in per_angle for s in sub]
|
||
all_claims = [c for s in all_sources for c in s["claims"]]
|
||
ranked = rank_claims(all_claims, config.MAX_VERIFY_CLAIMS)
|
||
|
||
def _sources_out():
|
||
return [{"url": s["url"], "quality": s["sourceQuality"], "angle": s["angle"],
|
||
"claimCount": len(s["claims"])} for s in all_sources]
|
||
|
||
if not ranked:
|
||
return {
|
||
"question": q,
|
||
"summary": f"No claims extracted. {len(all_sources)} sources fetched, all empty/failed.",
|
||
"findings": [], "refuted": [], "sources": _sources_out(),
|
||
"stats": {"angles": len(scope.angles), "sources": len(all_sources),
|
||
"claims": 0, "dupes": len(deduper.dupes)},
|
||
}
|
||
|
||
# ── Verify (배리어) ──
|
||
async def verify_claim(claim: dict) -> dict:
|
||
async def one_vote(v: int):
|
||
async with sem:
|
||
return await backend.run_agent(
|
||
prompts.verify_prompt(q, claim, v, config.VOTES_PER_CLAIM, config.REFUTATIONS_REQUIRED),
|
||
Verdict, label="v" + str(v) + ":" + claim["claim"][:40])
|
||
verdicts = await asyncio.gather(*[one_vote(v) for v in range(config.VOTES_PER_CLAIM)])
|
||
t = tally([vd.model_dump() if vd is not None else None for vd in verdicts],
|
||
config.VOTES_PER_CLAIM, config.REFUTATIONS_REQUIRED)
|
||
return {**claim, **t}
|
||
|
||
voted = await asyncio.gather(*[verify_claim(c) for c in ranked])
|
||
confirmed = [c for c in voted if c["survives"]]
|
||
killed = [c for c in voted if not c["survives"]]
|
||
|
||
def _refuted_out():
|
||
return [{"claim": c["claim"],
|
||
"vote": str(len(c["valid"]) - c["refutedVotes"]) + "-" + str(c["refutedVotes"]),
|
||
"source": c["sourceUrl"]} for c in killed]
|
||
|
||
if not confirmed:
|
||
return {
|
||
"question": q,
|
||
"summary": f"All {len(voted)} claims refuted by adversarial verification. Research inconclusive.",
|
||
"findings": [], "refuted": _refuted_out(), "sources": _sources_out(),
|
||
"stats": build_stats(angles=len(scope.angles), sources=len(all_sources),
|
||
claims=len(all_claims), voted=len(voted), confirmed=0,
|
||
killed=len(killed), after_synth=0, dupes=len(deduper.dupes),
|
||
budget_dropped=len(deduper.budget_dropped),
|
||
votes_per_claim=config.VOTES_PER_CLAIM),
|
||
}
|
||
|
||
# ── Synthesize ──
|
||
block, killed_block = build_synth_blocks(confirmed, killed)
|
||
report = await backend.run_agent(
|
||
prompts.synth_prompt(q, block, killed_block, len(confirmed), config.VOTES_PER_CLAIM),
|
||
Report, label="synthesize")
|
||
|
||
if report is None:
|
||
return {
|
||
"question": q,
|
||
"summary": f"Synthesis step was skipped or failed — returning {len(confirmed)} verified claims unmerged.",
|
||
"findings": [],
|
||
"confirmed": [{"claim": c["claim"], "source": c["sourceUrl"], "quote": c["quote"],
|
||
"vote": str(len(c["valid"]) - c["refutedVotes"]) + "-" + str(c["refutedVotes"])}
|
||
for c in confirmed],
|
||
"refuted": _refuted_out(), "sources": _sources_out(),
|
||
"stats": build_stats(angles=len(scope.angles), sources=len(all_sources),
|
||
claims=len(all_claims), voted=len(voted), confirmed=len(confirmed),
|
||
killed=len(killed), after_synth=0, dupes=len(deduper.dupes),
|
||
budget_dropped=len(deduper.budget_dropped),
|
||
votes_per_claim=config.VOTES_PER_CLAIM),
|
||
}
|
||
|
||
return {
|
||
"question": q,
|
||
**report.model_dump(),
|
||
"refuted": _refuted_out(), "sources": _sources_out(),
|
||
"stats": build_stats(angles=len(scope.angles), sources=len(all_sources),
|
||
claims=len(all_claims), voted=len(voted), confirmed=len(confirmed),
|
||
killed=len(killed), after_synth=len(report.findings),
|
||
dupes=len(deduper.dupes), budget_dropped=len(deduper.budget_dropped),
|
||
votes_per_claim=config.VOTES_PER_CLAIM),
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 4: 테스트 통과 확인**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/pytest tests/test_pipeline_mock.py -v`
|
||
Expected: PASS (5 passed)
|
||
|
||
- [ ] **Step 5: 전체 회귀**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/pytest -v`
|
||
Expected: PASS (전체)
|
||
|
||
- [ ] **Step 6: Commit**
|
||
|
||
```bash
|
||
git add scripts/deep-research/deep_research/pipeline.py scripts/deep-research/tests/test_pipeline_mock.py
|
||
git commit -m "feat(deep-research): asyncio pipeline orchestration (MockBackend TDD, 4 degenerate paths)"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 9: report.py (Report → markdown/JSON)
|
||
|
||
**Files:**
|
||
- Create: `scripts/deep-research/deep_research/report.py`
|
||
- Test: `scripts/deep-research/tests/test_pipeline_mock.py` (추가)
|
||
|
||
- [ ] **Step 1: 실패 테스트 작성** — `tests/test_pipeline_mock.py` 에 추가:
|
||
|
||
```python
|
||
from deep_research.report import to_markdown
|
||
|
||
|
||
def test_to_markdown_renders_summary_and_stats():
|
||
result = {
|
||
"question": "Q", "summary": "ans",
|
||
"findings": [{"claim": "F1", "confidence": "high", "sources": ["http://a"], "evidence": "e"}],
|
||
"caveats": "c", "refuted": [], "sources": [],
|
||
"stats": {"angles": 5, "confirmed": 1},
|
||
}
|
||
md = to_markdown(result)
|
||
assert "# Deep Research" in md and "ans" in md and "F1" in md
|
||
```
|
||
|
||
- [ ] **Step 2: 실패 확인**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/pytest tests/test_pipeline_mock.py -k to_markdown -v`
|
||
Expected: FAIL — `ModuleNotFoundError`
|
||
|
||
- [ ] **Step 3: report.py 구현**
|
||
|
||
```python
|
||
import json
|
||
|
||
|
||
def to_markdown(result: dict) -> str:
|
||
if "error" in result:
|
||
return "# Deep Research\n\n**Error:** " + result["error"]
|
||
|
||
lines = ["# Deep Research", "", "**Question:** " + result.get("question", ""), ""]
|
||
lines += ["## Summary", result.get("summary", ""), ""]
|
||
|
||
findings = result.get("findings", [])
|
||
if findings:
|
||
lines.append("## Findings")
|
||
for i, f in enumerate(findings, 1):
|
||
srcs = ", ".join(f.get("sources", []))
|
||
lines += [
|
||
f"### {i}. {f['claim']} _({f['confidence']})_",
|
||
f"{f.get('evidence', '')}",
|
||
f"Sources: {srcs}",
|
||
"",
|
||
]
|
||
|
||
caveats = result.get("caveats")
|
||
if caveats:
|
||
lines += ["## Caveats", caveats, ""]
|
||
|
||
oq = result.get("openQuestions")
|
||
if oq:
|
||
lines += ["## Open Questions"] + [f"- {q}" for q in oq] + [""]
|
||
|
||
refuted = result.get("refuted", [])
|
||
if refuted:
|
||
lines.append("## Refuted (transparency)")
|
||
for r in refuted:
|
||
lines.append(f"- \"{r['claim']}\" (vote {r.get('vote', '')}, {r.get('source', '')})")
|
||
lines.append("")
|
||
|
||
lines += ["## Stats", "```json", json.dumps(result.get("stats", {}), ensure_ascii=False, indent=2), "```"]
|
||
return "\n".join(lines)
|
||
|
||
|
||
def to_json(result: dict) -> str:
|
||
return json.dumps(result, ensure_ascii=False, indent=2)
|
||
```
|
||
|
||
- [ ] **Step 4: 통과 확인**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/pytest tests/test_pipeline_mock.py -k to_markdown -v`
|
||
Expected: PASS
|
||
|
||
- [ ] **Step 5: Commit**
|
||
|
||
```bash
|
||
git add scripts/deep-research/deep_research/report.py scripts/deep-research/tests/test_pipeline_mock.py
|
||
git commit -m "feat(deep-research): report renderer (markdown + json)"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 10: backends/codex.py (`codex exec --json --output-schema`)
|
||
|
||
**Files:**
|
||
- Create: `scripts/deep-research/deep_research/backends/codex.py`
|
||
- Test: `scripts/deep-research/tests/test_pipeline_mock.py` (파서 단위테스트 추가; subprocess 는 모킹)
|
||
|
||
> codex 의 JSONL stdout 에서 최종 agent 메시지(스키마 JSON)를 뽑는 파서를 분리해 단위테스트한다. 실제 `codex exec` 호출은 수동 스모크(Task 13).
|
||
|
||
- [ ] **Step 1: 파서 실패 테스트 작성** — `tests/test_pipeline_mock.py` 에 추가:
|
||
|
||
```python
|
||
from deep_research.backends.codex import extract_final_json
|
||
|
||
|
||
def test_extract_final_json_from_jsonl():
|
||
jsonl = "\n".join([
|
||
'{"type":"reasoning","text":"thinking"}',
|
||
'{"type":"web_search","query":"x"}',
|
||
'{"type":"agent_message","text":"{\\"question\\":\\"Q\\",\\"summary\\":\\"s\\",\\"angles\\":[]}"}',
|
||
])
|
||
obj = extract_final_json(jsonl)
|
||
assert obj["summary"] == "s"
|
||
|
||
|
||
def test_extract_final_json_returns_none_when_absent():
|
||
assert extract_final_json('{"type":"reasoning","text":"only"}') is None
|
||
```
|
||
|
||
- [ ] **Step 2: 실패 확인**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/pytest tests/test_pipeline_mock.py -k extract_final_json -v`
|
||
Expected: FAIL — `ModuleNotFoundError`
|
||
|
||
- [ ] **Step 3: codex.py 구현**
|
||
|
||
```python
|
||
"""Codex CLI headless backend. `codex exec --json --output-schema` subprocess."""
|
||
import asyncio
|
||
import json
|
||
import os
|
||
import tempfile
|
||
from pydantic import BaseModel, ValidationError
|
||
from deep_research.backends.base import AgentBackend
|
||
|
||
|
||
def extract_final_json(jsonl_stdout: str) -> dict | None:
|
||
"""JSONL 이벤트 스트림에서 마지막 agent_message 의 text(JSON) 파싱."""
|
||
final = None
|
||
for line in jsonl_stdout.splitlines():
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
evt = json.loads(line)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
if evt.get("type") in ("agent_message", "message") and "text" in evt:
|
||
final = evt["text"]
|
||
if final is None:
|
||
return None
|
||
try:
|
||
return json.loads(final)
|
||
except json.JSONDecodeError:
|
||
return None
|
||
|
||
|
||
class CodexBackend(AgentBackend):
|
||
def __init__(self, *, neutral_cwd: str | None = None, model: str | None = None):
|
||
self.neutral_cwd = neutral_cwd or tempfile.gettempdir()
|
||
self.model = model
|
||
|
||
async def run_agent(self, prompt: str, schema: type[BaseModel], *, label: str):
|
||
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as fh:
|
||
json.dump(schema.model_json_schema(), fh)
|
||
schema_path = fh.name
|
||
cmd = ["codex", "exec", "--json", "--output-schema", schema_path,
|
||
"--cd", self.neutral_cwd, "-s", "read-only"]
|
||
if self.model:
|
||
cmd += ["-m", self.model]
|
||
cmd.append(prompt)
|
||
try:
|
||
proc = await asyncio.create_subprocess_exec(
|
||
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
|
||
out, _ = await proc.communicate()
|
||
finally:
|
||
os.unlink(schema_path)
|
||
if proc.returncode != 0:
|
||
return None
|
||
obj = extract_final_json(out.decode("utf-8", "replace"))
|
||
if obj is None:
|
||
return None
|
||
try:
|
||
return schema.model_validate(obj)
|
||
except ValidationError:
|
||
return None
|
||
```
|
||
|
||
- [ ] **Step 4: 통과 확인**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/pytest tests/test_pipeline_mock.py -k extract_final_json -v`
|
||
Expected: PASS (2 passed)
|
||
|
||
- [ ] **Step 5: Commit**
|
||
|
||
```bash
|
||
git add scripts/deep-research/deep_research/backends/codex.py scripts/deep-research/tests/test_pipeline_mock.py
|
||
git commit -m "feat(deep-research): codex backend (codex exec --json --output-schema)"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 11: backends/antigravity.py (`agy -p` + 검증·재시도)
|
||
|
||
**Files:**
|
||
- Create: `scripts/deep-research/deep_research/backends/antigravity.py`
|
||
- Test: `scripts/deep-research/tests/test_pipeline_mock.py` (파서 단위테스트 추가)
|
||
|
||
> `agy` 의 output-schema 플래그는 미확인(spec §11) → 프롬프트에 스키마 JSON 을 첨부하고, 출력에서 JSON 블록을 추출·검증·재시도. JSON 추출기를 분리해 단위테스트.
|
||
|
||
- [ ] **Step 1: 추출기 실패 테스트 작성** — `tests/test_pipeline_mock.py` 에 추가:
|
||
|
||
```python
|
||
from deep_research.backends.antigravity import extract_json_block
|
||
|
||
|
||
def test_extract_json_block_from_fenced():
|
||
text = 'prelude\n```json\n{"refuted": false, "evidence": "e", "confidence": "high"}\n```\ntrailing'
|
||
obj = extract_json_block(text)
|
||
assert obj["confidence"] == "high"
|
||
|
||
|
||
def test_extract_json_block_bare_object():
|
||
obj = extract_json_block('noise {"refuted": true, "evidence": "e", "confidence": "low"} more')
|
||
assert obj["refuted"] is True
|
||
|
||
|
||
def test_extract_json_block_none_when_absent():
|
||
assert extract_json_block("no json here") is None
|
||
```
|
||
|
||
- [ ] **Step 2: 실패 확인**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/pytest tests/test_pipeline_mock.py -k extract_json_block -v`
|
||
Expected: FAIL — `ModuleNotFoundError`
|
||
|
||
- [ ] **Step 3: antigravity.py 구현**
|
||
|
||
```python
|
||
"""Antigravity CLI headless backend. `agy -p` subprocess + 드라이버 스키마 검증·재시도."""
|
||
import asyncio
|
||
import json
|
||
import re
|
||
import tempfile
|
||
from pydantic import BaseModel, ValidationError
|
||
from deep_research.backends.base import AgentBackend
|
||
|
||
_FENCE = re.compile(r"```(?:json)?\s*(\{.*?\})\s*```", re.DOTALL)
|
||
|
||
|
||
def extract_json_block(text: str) -> dict | None:
|
||
"""출력 텍스트에서 JSON 객체 추출: 펜스 우선, 없으면 최외곽 중괄호."""
|
||
m = _FENCE.search(text)
|
||
candidates = [m.group(1)] if m else []
|
||
if not candidates:
|
||
start, depth = -1, 0
|
||
for i, ch in enumerate(text):
|
||
if ch == "{":
|
||
if depth == 0:
|
||
start = i
|
||
depth += 1
|
||
elif ch == "}" and depth > 0:
|
||
depth -= 1
|
||
if depth == 0 and start >= 0:
|
||
candidates.append(text[start:i + 1])
|
||
break
|
||
for c in candidates:
|
||
try:
|
||
return json.loads(c)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
return None
|
||
|
||
|
||
class AntigravityBackend(AgentBackend):
|
||
def __init__(self, *, retries: int = 2, neutral_cwd: str | None = None, model: str | None = None):
|
||
self.retries = retries
|
||
self.neutral_cwd = neutral_cwd or tempfile.gettempdir()
|
||
self.model = model
|
||
|
||
async def _invoke(self, prompt: str) -> str | None:
|
||
cmd = ["agy", "-p", prompt, "--cd", self.neutral_cwd]
|
||
if self.model:
|
||
cmd += ["-m", self.model]
|
||
proc = await asyncio.create_subprocess_exec(
|
||
*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
|
||
out, _ = await proc.communicate()
|
||
if proc.returncode != 0:
|
||
return None
|
||
return out.decode("utf-8", "replace")
|
||
|
||
async def run_agent(self, prompt: str, schema: type[BaseModel], *, label: str):
|
||
schema_json = json.dumps(schema.model_json_schema())
|
||
full = (prompt + "\n\n## Output format\nReturn ONLY a JSON object conforming to this "
|
||
"JSON Schema (no prose, no markdown fences):\n" + schema_json)
|
||
for attempt in range(self.retries + 1):
|
||
text = await self._invoke(full if attempt == 0 else
|
||
full + "\n\nYour previous output was invalid. Return ONLY the JSON object.")
|
||
if text is None:
|
||
continue
|
||
obj = extract_json_block(text)
|
||
if obj is None:
|
||
continue
|
||
try:
|
||
return schema.model_validate(obj)
|
||
except ValidationError:
|
||
continue
|
||
return None
|
||
```
|
||
|
||
- [ ] **Step 4: 통과 확인**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/pytest tests/test_pipeline_mock.py -k extract_json_block -v`
|
||
Expected: PASS (3 passed)
|
||
|
||
- [ ] **Step 5: Commit**
|
||
|
||
```bash
|
||
git add scripts/deep-research/deep_research/backends/antigravity.py scripts/deep-research/tests/test_pipeline_mock.py
|
||
git commit -m "feat(deep-research): antigravity backend (agy -p + validate/retry)"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 12: __main__.py + README
|
||
|
||
**Files:**
|
||
- Create: `scripts/deep-research/deep_research/__main__.py`
|
||
- Create: `scripts/deep-research/README.md`
|
||
|
||
- [ ] **Step 1: __main__.py 구현**
|
||
|
||
```python
|
||
"""CLI: python -m deep_research --backend {codex|antigravity} "<질문>" [--json]"""
|
||
import argparse
|
||
import asyncio
|
||
import shutil
|
||
import sys
|
||
from deep_research.config import DEFAULT
|
||
from deep_research.pipeline import run_research
|
||
from deep_research.report import to_markdown, to_json
|
||
|
||
|
||
def _make_backend(name: str):
|
||
if name == "codex":
|
||
if shutil.which("codex") is None:
|
||
sys.exit("error: `codex` CLI not found on PATH. Install + login (subscription).")
|
||
from deep_research.backends.codex import CodexBackend
|
||
return CodexBackend()
|
||
if name == "antigravity":
|
||
if shutil.which("agy") is None:
|
||
sys.exit("error: `agy` CLI not found on PATH. Install + login (subscription).")
|
||
from deep_research.backends.antigravity import AntigravityBackend
|
||
return AntigravityBackend(retries=DEFAULT.ANTIGRAVITY_RETRIES)
|
||
sys.exit(f"error: unknown backend '{name}'")
|
||
|
||
|
||
def main(argv=None):
|
||
ap = argparse.ArgumentParser(prog="deep_research")
|
||
ap.add_argument("question")
|
||
ap.add_argument("--backend", required=True, choices=["codex", "antigravity"])
|
||
ap.add_argument("--json", action="store_true", help="emit raw JSON instead of markdown")
|
||
ap.add_argument("--concurrency", type=int, default=DEFAULT.CONCURRENCY)
|
||
args = ap.parse_args(argv)
|
||
|
||
backend = _make_backend(args.backend)
|
||
cfg = DEFAULT.__class__(CONCURRENCY=args.concurrency)
|
||
result = asyncio.run(run_research(args.question, backend, config=cfg))
|
||
print(to_json(result) if args.json else to_markdown(result))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
```
|
||
|
||
- [ ] **Step 2: README.md 작성**
|
||
|
||
```markdown
|
||
# deep-research driver
|
||
|
||
Claude Code 의 deep-research 하네스를 **codex-cli / antigravity-cli** 로 이식한 외부 Python 드라이버.
|
||
|
||
## 전제
|
||
- `codex` 또는 `agy` CLI 설치 + 로그인(구독). 별도 검색 API 키 불필요 — 웹 조사는 각 CLI 네이티브 도구가 수행.
|
||
|
||
## 사용
|
||
```bash
|
||
python -m deep_research --backend codex "2026년 한국 전기차 보조금 정책 변화"
|
||
python -m deep_research --backend antigravity "..." --json
|
||
```
|
||
|
||
## 구조
|
||
- `core.py` — 결정론 제어(dedup·예산·3-vote 정족수·랭킹). 원본 JS 와 1:1, 순수함수.
|
||
- `backends/` — 플랫폼 어댑터(codex/antigravity/mock).
|
||
- 충실도 기준 원본: `deep-research-wf_aecef33f-4cc.js` (spec 참조).
|
||
|
||
## 테스트
|
||
```bash
|
||
pip install -e ".[dev]" && pytest
|
||
```
|
||
MockBackend 로 네트워크 0 상태에서 6페이즈 + 퇴화 경로 3종 검증.
|
||
```
|
||
|
||
- [ ] **Step 3: CLI smoke (인자 파싱만, CLI 미설치 시 안내 확인)**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/python -m deep_research --backend codex "test" 2>&1 | head -1` (codex 미설치면 안내 메시지)
|
||
Expected: 인자 파싱 동작 — codex 있으면 실행, 없으면 `error: codex CLI not found`.
|
||
|
||
- [ ] **Step 4: 전체 테스트 회귀**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/pytest -v`
|
||
Expected: PASS (전체)
|
||
|
||
- [ ] **Step 5: Commit**
|
||
|
||
```bash
|
||
git add scripts/deep-research/deep_research/__main__.py scripts/deep-research/README.md
|
||
git commit -m "feat(deep-research): CLI entrypoint + README"
|
||
```
|
||
|
||
---
|
||
|
||
### Task 13: 실 backend 스모크 (수동, 선택)
|
||
|
||
**Files:** 없음 (실행 검증만)
|
||
|
||
- [ ] **Step 1: codex 실 스모크**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/python -m deep_research --backend codex "What are the main tradeoffs of virtual threads vs reactive programming in Java?"`
|
||
Expected: 보고서 markdown + stats. `agentCalls` 가 `1+angles+sources+voted*3+1` 와 부합. web_search 가 실제로 돌았는지 codex 로그로 확인.
|
||
|
||
- [ ] **Step 2: antigravity 실 스모크**
|
||
|
||
Run: `cd scripts/deep-research && .venv/bin/python -m deep_research --backend antigravity "<같은 질문>"`
|
||
Expected: 보고서 출력. 스키마 검증 실패율 높으면 `--output format` 플래그 등 `agy` 옵션 조사(spec §11 리스크), antigravity.py 재프롬프트 동작 확인.
|
||
|
||
- [ ] **Step 3: 관찰 기록**
|
||
|
||
두 backend 산출물 차이(스키마 준수율, 속도, 웹도구 동작)를 `docs/superpowers/notes/2026-06-09-deep-research-port-smoke.md` 에 메모(Write 도구 사용).
|
||
|
||
---
|
||
|
||
## Self-Review (작성자 체크)
|
||
|
||
**1. Spec coverage**
|
||
- §3 모듈 구조 → Task 0~12 전부 매핑 ✓
|
||
- §4 상수 → Task 1 ✓ · §5 스키마 → Task 2 ✓ · §6 6페이즈 → Task 3(프롬프트)+Task 8(흐름) ✓
|
||
- §6.2 dedup 수학 → Task 4 ✓ · §6.4 rank → Task 5 ✓ · §6.5 정족수 abstention 엣지 → Task 5 ✓
|
||
- §6.7 퇴화 경로 3종 → Task 8 테스트 ✓ · §6.8 stats → Task 6 ✓
|
||
- §7 동시성(무배리어/배리어/dedup 락) → Task 8 ✓
|
||
- §10 테스트(core 단위 + MockBackend 통합) → Task 4~8 ✓
|
||
- backends(codex/antigravity/mock) → Task 7/10/11 ✓
|
||
- §1.3 CLI headless + 네이티브 웹 → Task 10/11 ✓ (web 계층 없음 일관)
|
||
|
||
**2. Placeholder scan** — TBD/TODO 없음. 모든 코드 스텝에 실제 코드 포함 ✓
|
||
|
||
**3. Type consistency**
|
||
- `Config` 필드명 Task1↔전 태스크 일치 ✓
|
||
- core dict 키 `sourceUrl`/`sourceQuality`/`refutedVotes`/`valid`/`survives` 일관(Task5 tally 출력 ↔ Task6 build_synth_blocks 입력 ↔ Task8 pipeline) ✓
|
||
- `host_of`(Task4) ↔ pipeline fetch label(Task8) 일치 ✓
|
||
- backend `run_agent(prompt, schema, *, label)` 시그니처 base↔mock↔codex↔antigravity 일치 ✓
|
||
- 스키마 camelCase 필드(`sourceQuality`/`publishDate`/`counterSource`/`openQuestions`) ↔ 프롬프트 지시 일치 ✓
|