218 lines
9.3 KiB
Python
218 lines
9.3 KiB
Python
"""extract.py — Tiered Extraction 브로커: bulk 발췌를 외부 구독 CLI 에 위임 + 결정론 검증.
|
||
|
||
설계 (rules/extraction-tiering.md):
|
||
- 드라이버가 파일을 *직접 읽어* 줄번호 매긴 본문을 프롬프트에 내장 — 외부 엔진은
|
||
repo 접근이 불필요하다 (쓰기 위험 0 / 입력 결정론 / codex·agy·mock 동일 동작).
|
||
- 외부 발췌는 신뢰하지 않는다: 모든 verbatim 인용을 원문에 re-match 해 검증/정정/폐기
|
||
(quote-verifier 내장). 통과분만 digest 로 방출 — 상위 티어(opus)는 digest 만 소비.
|
||
- engine funnel: 어떤 엔진이 몇 파일을 처리/실패했는지 항상 기록 (no silent engine swap).
|
||
|
||
CLI:
|
||
python3 -m deep_research.extract --backend codex|antigravity|auto \
|
||
--question "<연구 질문>" --files f1.md f2.md ... [--out digest.md] [--max-quotes 8]
|
||
exit: 0 (digest 생성, 부분 실패는 funnel 에 기록) / 1 (전 파일 실패) / 2 (인자 오류)
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
from pydantic import BaseModel, Field
|
||
|
||
MAX_FILE_CHARS = 60_000 # 파일당 프롬프트 상한 (초과분은 절단 + funnel 기록)
|
||
CONCURRENCY = 3
|
||
|
||
|
||
class Quote(BaseModel):
|
||
line: int = Field(description="1-based line number in the file")
|
||
quote: str = Field(description="byte-for-byte verbatim quote from that line (no paraphrase)")
|
||
|
||
|
||
class FileExtraction(BaseModel):
|
||
relevant: bool = Field(description="whether this file contains material relevant to the question")
|
||
summary: str = Field(description="2-4 sentence summary of what this file says about the question")
|
||
facts: list[str] = Field(default_factory=list,
|
||
description="key facts relevant to the question, each one sentence")
|
||
quotes: list[Quote] = Field(default_factory=list,
|
||
description="supporting verbatim quotes with line numbers")
|
||
|
||
|
||
PROMPT = """You are a read-only extraction worker. Below is ONE file with line numbers, \
|
||
and a research question. Extract ONLY what the file actually says — no inference, no outside \
|
||
knowledge, no paraphrase inside quotes.
|
||
|
||
Research question: {question}
|
||
|
||
Rules:
|
||
- quotes must be byte-for-byte substrings of a single line (copy exactly, without the line-number prefix).
|
||
- give the 1-based line number for each quote.
|
||
- at most {max_quotes} quotes; prefer the most decision-relevant lines.
|
||
- if the file is irrelevant to the question, set relevant=false with a one-line summary.
|
||
- respond ONLY with the JSON object matching the schema.
|
||
|
||
FILE: {path}
|
||
----------------------------------------
|
||
{numbered}
|
||
----------------------------------------"""
|
||
|
||
|
||
def numbered_content(text: str, limit: int = MAX_FILE_CHARS) -> tuple[str, bool]:
|
||
lines = text.splitlines()
|
||
out, total, truncated = [], 0, False
|
||
for i, line in enumerate(lines, start=1):
|
||
s = f"{i}\t{line}"
|
||
total += len(s) + 1
|
||
if total > limit:
|
||
truncated = True
|
||
break
|
||
out.append(s)
|
||
return "\n".join(out), truncated
|
||
|
||
|
||
def verify_quotes(path: Path, ext: FileExtraction) -> dict:
|
||
"""결정론 quote-verifier: verbatim 재대조. 반환: 검증 통계 + 정정된 quotes.
|
||
|
||
판정: 인용이 명시 라인에 있으면 PASS / 다른 라인에 있으면 CORRECTED(라인 정정) /
|
||
어디에도 없으면 DROPPED (위조·의역 — 폐기)."""
|
||
try:
|
||
lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
|
||
except Exception:
|
||
return {"pass": 0, "corrected": 0, "dropped": len(ext.quotes), "kept": []}
|
||
n_pass = n_corr = n_drop = 0
|
||
kept: list[Quote] = []
|
||
for q in ext.quotes:
|
||
quote = q.quote.strip()
|
||
if not quote:
|
||
n_drop += 1
|
||
continue
|
||
if 1 <= q.line <= len(lines) and quote in lines[q.line - 1]:
|
||
n_pass += 1
|
||
kept.append(q)
|
||
continue
|
||
hit = next((i for i, l in enumerate(lines, start=1) if quote in l), None)
|
||
if hit is not None:
|
||
n_corr += 1
|
||
kept.append(Quote(line=hit, quote=quote))
|
||
else:
|
||
n_drop += 1
|
||
return {"pass": n_pass, "corrected": n_corr, "dropped": n_drop, "kept": kept}
|
||
|
||
|
||
def make_backend(name: str):
|
||
if name == "codex":
|
||
from deep_research.backends.codex import CodexBackend
|
||
return CodexBackend()
|
||
if name == "antigravity":
|
||
from deep_research.backends.antigravity import AntigravityBackend
|
||
return AntigravityBackend()
|
||
raise ValueError(name)
|
||
|
||
|
||
async def extract_file(backends: list[tuple[str, object]], question: str, path: Path,
|
||
max_quotes: int, sem: asyncio.Semaphore) -> dict:
|
||
"""파일 1개 발췌 — fallback 사다리 순서대로 시도. 반환 dict 는 digest 렌더 입력."""
|
||
try:
|
||
text = path.read_text(encoding="utf-8", errors="replace")
|
||
except Exception as e:
|
||
return {"path": str(path), "engine": None, "error": f"read 실패: {e}"}
|
||
numbered, truncated = numbered_content(text)
|
||
prompt = PROMPT.format(question=question, max_quotes=max_quotes,
|
||
path=path, numbered=numbered)
|
||
async with sem:
|
||
for engine_name, backend in backends:
|
||
result = await backend.run_agent(prompt, FileExtraction, label=f"extract:{path.name}")
|
||
if result is not None:
|
||
v = verify_quotes(path, result)
|
||
return {"path": str(path), "engine": engine_name, "truncated": truncated,
|
||
"relevant": result.relevant, "summary": result.summary,
|
||
"facts": result.facts, "verify": v}
|
||
return {"path": str(path), "engine": None, "error": "전 엔진 실패 (가용성/스키마)"}
|
||
|
||
|
||
def render_digest(question: str, results: list[dict]) -> str:
|
||
ok = [r for r in results if "error" not in r]
|
||
failed = [r for r in results if "error" in r]
|
||
engines: dict[str, int] = {}
|
||
for r in ok:
|
||
engines[r["engine"]] = engines.get(r["engine"], 0) + 1
|
||
out = [f"# Extraction Digest", f"**Question:** {question}",
|
||
f"**Files:** {len(results)} (성공 {len(ok)} / 실패 {len(failed)})",
|
||
f"**Engines:** " + (", ".join(f"{k}×{v}" for k, v in engines.items()) or "없음"), ""]
|
||
for r in ok:
|
||
v = r["verify"]
|
||
out.append(f"## {r['path']}" + (" (TRUNCATED)" if r.get("truncated") else ""))
|
||
if not r["relevant"]:
|
||
out.append(f"- 무관: {r['summary']}")
|
||
out.append("")
|
||
continue
|
||
out.append(f"- 요약: {r['summary']}")
|
||
for f in r["facts"]:
|
||
out.append(f"- {f}")
|
||
for q in v["kept"]:
|
||
out.append(f" > \"{q.quote}\" — {r['path']}:{q.line}")
|
||
out.append(f"- 인용 검증: PASS {v['pass']} / 정정 {v['corrected']} / **폐기 {v['dropped']}**")
|
||
if r["facts"] and not v["kept"]:
|
||
# 검증 통과 인용이 0건이면 위 facts 는 근거 없는 주장 (계명 2 — 무검증 발췌 소비 금지)
|
||
out.append("- ⚠ 검증 인용 0건 — 위 facts 는 미검증 주장 (haiku/sonnet 재발췌 후보)")
|
||
out.append("")
|
||
if failed:
|
||
out.append("## 실패 (haiku/sonnet 재발췌 대상)")
|
||
for r in failed:
|
||
out.append(f"- {r['path']}: {r['error']}")
|
||
out.append("")
|
||
total_q = sum(r["verify"]["pass"] + r["verify"]["corrected"] + r["verify"]["dropped"] for r in ok)
|
||
kept_q = sum(len(r["verify"]["kept"]) for r in ok)
|
||
out.append("```wiki-stats")
|
||
out.append("agent: extract-broker")
|
||
out.append(f"found: {len(results)}")
|
||
out.append(f"processed: {len(ok)}")
|
||
out.append(f"dropped: {len(failed)}")
|
||
if failed:
|
||
out.append(f"dropped_reason: 엔진 실패 {len(failed)}건 (위 실패 목록 — 상위 티어 재발췌)")
|
||
out.append("```")
|
||
out.append(f"<!-- quotes: total {total_q}, verified-kept {kept_q} -->")
|
||
return "\n".join(out)
|
||
|
||
|
||
async def run(args) -> int:
|
||
order = {"codex": ["codex", "antigravity"],
|
||
"antigravity": ["antigravity", "codex"],
|
||
"auto": ["codex", "antigravity"]}[args.backend]
|
||
backends = []
|
||
for name in order:
|
||
try:
|
||
backends.append((name, make_backend(name)))
|
||
except Exception:
|
||
continue
|
||
if not backends:
|
||
print("error: 사용 가능한 backend 없음", file=sys.stderr)
|
||
return 1
|
||
sem = asyncio.Semaphore(CONCURRENCY)
|
||
results = await asyncio.gather(*[
|
||
extract_file(backends, args.question, Path(f), args.max_quotes, sem)
|
||
for f in args.files])
|
||
digest = render_digest(args.question, list(results))
|
||
if args.out:
|
||
Path(args.out).write_text(digest, encoding="utf-8")
|
||
print(f"digest → {args.out} ({len(digest)} chars)")
|
||
else:
|
||
print(digest)
|
||
return 0 if any("error" not in r for r in results) else 1
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser(description="Tiered extraction 브로커 (외부 구독 CLI + 결정론 검증)")
|
||
ap.add_argument("--backend", default="auto", choices=["codex", "antigravity", "auto"])
|
||
ap.add_argument("--question", required=True)
|
||
ap.add_argument("--files", nargs="+", required=True)
|
||
ap.add_argument("--out", help="digest 출력 파일 (생략 시 stdout)")
|
||
ap.add_argument("--max-quotes", type=int, default=8)
|
||
args = ap.parse_args()
|
||
sys.exit(asyncio.run(run(args)))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|