byeongwook.choiandClaude Fable 5.1
0f70c0d245
- settings: LLM_PROVIDER/LLM_CHAT_PATH/LLM_BODY_MODEL/LLM_JSON_MODE/LLM_MAX_TOKENS, FABRIX_* 3종, llm_enabled() - llm_client: headers()/body() 를 규격별로 구성. fabrix 는 x-openapi-token(Bearer)/x-generative-ai-client/ x-llm-model-id/x-generative-ai-user-email, body model 은 LLM_BODY_MODEL - summarize/ping: 접속 점검 명령 (--show 로 요청만 확인) - docs/llm-provider-plan.md: 계획·.env 값·오류별 조치. 기본값은 openai 라 기존 동작 불변 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
429 lines
17 KiB
Python
429 lines
17 KiB
Python
"""Stage 5 — 질의 API (FastAPI).
|
|
|
|
opencode-be 의 커스텀 툴(.opencode/tool/_index_client.ts)이 호출하는 HTTP 서비스.
|
|
실행: python -m query.api (기본 127.0.0.1:8100)
|
|
자연어 로직 질의는 /search/logic (로직 조각) 이 1차 진입점이다 — docs/logic-chunk-design.md
|
|
|
|
응답 크기 원칙: 툴 응답이 LLM 컨텍스트에 그대로 들어가므로 상한을 두고 자른다.
|
|
(책임 경계 — 컨텍스트 초과는 인덱스 쪽 책임: 계획서 리스크 항목)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
from datetime import datetime, timezone
|
|
|
|
import threading
|
|
|
|
from fastapi import BackgroundTasks, Body, FastAPI, HTTPException, Query
|
|
from fastapi.responses import HTMLResponse
|
|
|
|
from config.settings import ROOT, settings
|
|
from index.db import connect, db_path
|
|
from index.loader import load_parsed, refresh_program_fts
|
|
from ingest.normalize import write_program
|
|
from parser.run import parse_program
|
|
from summarize.runner import PROMPT_VERSION, summarize_units
|
|
|
|
from . import tools
|
|
from .dashboard import render_dashboard
|
|
|
|
app = FastAPI(title="abap-indexing", version="0.1.0")
|
|
|
|
MAX_CODE_LINES = 400 # unit 코드 응답 상한 (초과 시 write 지점 위주로 잘라야 하나 v1은 앞부분+안내)
|
|
|
|
|
|
def _wrap(fn, *args, **kwargs):
|
|
try:
|
|
return fn(*args, **kwargs)
|
|
except tools.NotFound as e:
|
|
raise HTTPException(status_code=404, detail=str(e)) from e
|
|
except sqlite3.OperationalError as e:
|
|
raise HTTPException(status_code=503, detail=f"인덱스 DB 오류: {e}") from e
|
|
|
|
|
|
# 프로세스 시작 시각 — 코드를 고쳤는데 반영이 안 될 때 서버가 낡았는지 판단하는 근거.
|
|
# (템플릿 html 은 요청마다 다시 읽지만 파이썬 모듈은 sys.modules 에 캐시된다.)
|
|
STARTED_AT = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
|
|
|
|
@app.get("/health")
|
|
def health() -> dict:
|
|
con = connect()
|
|
try:
|
|
programs = con.execute("SELECT COUNT(*) AS c FROM program").fetchone()["c"]
|
|
with_source = con.execute("SELECT COUNT(*) AS c FROM program WHERE has_source=1").fetchone()["c"]
|
|
units = con.execute("SELECT COUNT(*) AS c FROM unit").fetchone()["c"]
|
|
chunks = con.execute("SELECT COUNT(*) AS c FROM logic_chunk").fetchone()["c"]
|
|
wiki_programs = (
|
|
sum(1 for _ in (settings.wiki_dir / "programs").glob("*.md"))
|
|
if (settings.wiki_dir / "programs").exists() else 0
|
|
)
|
|
return {
|
|
"status": "ok",
|
|
"started_at": STARTED_AT,
|
|
"db": str(db_path()),
|
|
"programs": programs,
|
|
"programs_with_source": with_source,
|
|
"units": units,
|
|
"logic_chunks": chunks,
|
|
"wiki_programs": wiki_programs,
|
|
"wiki_search_backend": settings.wiki_search_backend, # qmd | pg (M3.5 파일럿으로 확정)
|
|
}
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
# 인제스트 후 백그라운드 요약 — 같은 프로그램 중복 실행 방지
|
|
_summarizing: set[str] = set()
|
|
_summarizing_lock = threading.Lock()
|
|
|
|
|
|
def _summarize_bg(program: str) -> None:
|
|
try:
|
|
summarize_units(program, limit=None, dry_run=False, trigger="ingest",
|
|
backend=settings.summarize_backend)
|
|
except Exception as e: # noqa: BLE001
|
|
print(f"[summarize:{program}] {type(e).__name__}: {e}")
|
|
finally:
|
|
with _summarizing_lock:
|
|
_summarizing.discard(program)
|
|
# 요약 후 위키 재생성 (계획서 v4 §5.7 — 실패해도 요약 결과에는 영향 없음)
|
|
try:
|
|
from wiki_out.okf_writer import write_program_wiki
|
|
|
|
write_program_wiki(program)
|
|
except Exception as e: # noqa: BLE001
|
|
print(f"[wiki:{program}] {type(e).__name__}: {e}")
|
|
|
|
|
|
@app.post("/ingest")
|
|
def ingest(payload: dict = Body(), background_tasks: BackgroundTasks = None) -> dict:
|
|
"""수집 API/gateway가 넘겨준 프로그램 소스 JSON 한 건을 정규화→파싱→적재까지 처리한다.
|
|
|
|
body 는 수집 원본과 같은 형태: {MAIN_PROGRAM, DESCRIPTION, INCLUDE_PROGRAM[], TEXT_SYMBOL?}
|
|
(gateway 를 거친 응답은 이미 strict JSON 이므로 raw 정규화 단계는 불필요)
|
|
"""
|
|
if not isinstance(payload, dict) or not payload.get("MAIN_PROGRAM"):
|
|
raise HTTPException(status_code=400, detail="MAIN_PROGRAM이 포함된 소스 JSON이 필요합니다")
|
|
|
|
try:
|
|
meta = write_program(payload, settings.data_normalized)
|
|
program = meta["program"]
|
|
parsed = parse_program(settings.data_normalized / program)
|
|
settings.data_parsed.mkdir(parents=True, exist_ok=True)
|
|
parse_path = settings.data_parsed / f"{program}.parse.json"
|
|
parse_path.write_text(json.dumps(parsed, ensure_ascii=False), encoding="utf-8")
|
|
except Exception as e: # noqa: BLE001
|
|
raise HTTPException(status_code=422, detail=f"정규화/파싱 실패: {type(e).__name__}: {e}") from e
|
|
|
|
con = connect()
|
|
try:
|
|
status = load_parsed(con, parse_path)
|
|
# 단건만 갱신 — 전량 재구축을 /ingest 마다 하면 1만 본에서 O(N²) 가 된다
|
|
refresh_program_fts(con, program)
|
|
con.commit()
|
|
# 요약이 안 됐거나(failed 포함) 프롬프트 버전이 지난 unit 수 — runner 의 스킵 조건과 동일 기준
|
|
pending = con.execute(
|
|
"SELECT COUNT(*) AS c FROM unit WHERE program=? AND unit_type IN "
|
|
"('FORM','METHOD','FUNCTION','MODULE','EVENT') "
|
|
"AND (summary_status!='done' OR prompt_version!=?)",
|
|
(program, PROMPT_VERSION),
|
|
).fetchone()["c"]
|
|
except Exception as e: # noqa: BLE001
|
|
con.rollback()
|
|
raise HTTPException(status_code=500, detail=f"인덱스 적재 실패: {type(e).__name__}: {e}") from e
|
|
finally:
|
|
con.close()
|
|
|
|
# LLM 요약을 백그라운드로 이어 돌린다 (응답은 즉시 반환).
|
|
# 소스가 안 바뀐(skip) 프로그램이라도 요약 미완료 unit 이 있으면 이어서 돌린다(재인덱싱 = 요약 재시도).
|
|
# SUMMARIZE_BACKEND=file 이면 키 없이 프롬프트만 큐에 쌓는다 (/summaries/jobs 로 확인).
|
|
backend = settings.summarize_backend
|
|
summarize = "disabled"
|
|
if backend != "off" and (backend == "file" or settings.llm_enabled()):
|
|
if status == "loaded" or pending:
|
|
with _summarizing_lock:
|
|
already = program in _summarizing
|
|
if not already:
|
|
_summarizing.add(program)
|
|
if already:
|
|
summarize = "already_running"
|
|
else:
|
|
background_tasks.add_task(_summarize_bg, program)
|
|
summarize = "scheduled" if status == "loaded" else "resumed"
|
|
else:
|
|
summarize = "skip" # 소스 동일 + 요약도 전부 완료
|
|
|
|
return {
|
|
"program": program,
|
|
"status": status, # loaded | skip(소스 변경 없음)
|
|
"summarize": summarize, # scheduled | resumed(미완료 요약 이어서) | already_running | skip | disabled
|
|
"summarize_backend": backend, # api | file(프롬프트 큐) | off
|
|
"pending_units": pending, # 요약 대기/실패 unit 수
|
|
"includes": len(meta["includes"]),
|
|
"units": parsed["stats"]["units"],
|
|
"statements": parsed["stats"]["statements"],
|
|
"unknown_ratio": parsed["stats"]["unknown_ratio"],
|
|
}
|
|
|
|
|
|
@app.get("/search/programs")
|
|
def search_programs(q: str = Query(min_length=1), top_k: int = Query(default=10, le=30)) -> dict:
|
|
return {"query": q, "results": _wrap(tools.search_programs, q, top_k)}
|
|
|
|
|
|
@app.get("/search/units")
|
|
def search_units(q: str = Query(min_length=1), program: str | None = None,
|
|
top_k: int = Query(default=10, le=30)) -> dict:
|
|
return {"query": q, "results": _wrap(tools.search_units, q, program, top_k)}
|
|
|
|
|
|
@app.get("/search/logic")
|
|
def search_logic(q: str = Query(min_length=1), program: str | None = None, kind: str | None = None,
|
|
top_k: int = Query(default=10, le=30)) -> dict:
|
|
"""로직 조각 검색 — 자연어 로직 질의의 1차 진입점. 프로그램 단위로 묶어 반환.
|
|
|
|
응답의 `expansion` 은 용어 사전으로 펼친 동의어를 보여준다. 조각마다 붙는 `matched_by` 가
|
|
'동의어 확장' 이면 원질의가 아니라 동의어로 걸린 결과다 (점수에 감쇠가 적용됨).
|
|
"""
|
|
return {"query": q, **_wrap(tools.search_logic, q, top_k, program, kind)}
|
|
|
|
|
|
@app.get("/chunks/{chunk_id}")
|
|
def chunk(chunk_id: str, decls: bool = True) -> dict:
|
|
"""조각 메타 + 코드 원문 + 정의부. chunk_id 의 '#' 은 URL 인코딩(%23) 해야 한다.
|
|
|
|
`code` 는 로직만이라 그대로 붙여넣으면 선언이 없어 문법 오류가 난다. 함께 가야 하는 선언은
|
|
`declaration_code`(붙여넣기용 한 덩어리)와 `declarations`(건별 목록)로 준다.
|
|
필요 없으면 `?decls=false` 로 끈다.
|
|
"""
|
|
return _wrap(tools.get_chunk, chunk_id, decls)
|
|
|
|
|
|
@app.get("/programs/{name}/declarations")
|
|
def program_declarations(name: str, scope: str | None = None) -> dict:
|
|
"""프로그램의 정의부 전체 (scope=global|unit 로 거를 수 있다)."""
|
|
return _wrap(tools.get_declarations, name, scope)
|
|
|
|
|
|
@app.get("/programs/{name}/chunks")
|
|
def program_chunks(name: str, unit_id: str | None = None) -> dict:
|
|
return {"program": name.upper(), "chunks": _wrap(tools.list_chunks, name, unit_id)}
|
|
|
|
|
|
@app.get("/programs/{name}/summary")
|
|
def program_summary(name: str) -> dict:
|
|
return _wrap(tools.get_program_summary, name)
|
|
|
|
|
|
@app.get("/programs/{name}/source")
|
|
def program_source(name: str) -> dict:
|
|
return _wrap(tools.get_program_source, name)
|
|
|
|
|
|
@app.get("/programs/{name}/units/{unit}/code")
|
|
def unit_code(name: str, unit: str) -> dict:
|
|
result = _wrap(tools.get_unit_code, name, unit)
|
|
lines = result["code"].split("\n")
|
|
if len(lines) > MAX_CODE_LINES:
|
|
result["code"] = "\n".join(lines[:MAX_CODE_LINES])
|
|
result["truncated"] = True
|
|
result["total_lines"] = len(lines)
|
|
return result
|
|
|
|
|
|
@app.get("/programs/{name}/trace/{symbol}")
|
|
def trace(name: str, symbol: str) -> dict:
|
|
return _wrap(tools.trace_variable, name, symbol)
|
|
|
|
|
|
@app.get("/programs/{name}/call-graph")
|
|
def call_graph(name: str, unit: str | None = None) -> dict:
|
|
return _wrap(tools.get_call_graph, name, unit)
|
|
|
|
|
|
@app.get("/programs/{name}/who-calls/{unit}")
|
|
def who_calls_route(name: str, unit: str) -> dict:
|
|
return _wrap(tools.who_calls, name, unit)
|
|
|
|
|
|
@app.get("/tables/{name}/usage")
|
|
def table_usage(name: str) -> dict:
|
|
return _wrap(tools.get_table_usage, name)
|
|
|
|
|
|
@app.get("/wiki/{doc_path:path}")
|
|
def wiki_doc(doc_path: str) -> dict:
|
|
"""위키 문서 원문 반환 (계획서 v4 §7.2).
|
|
|
|
경로 규약: **wiki/ 루트 상대 경로** (예: programs/ZFIR10070.md).
|
|
경로 A(qmd) 채택 시 qmd 검색 결과의 경로가 그대로 들어온다.
|
|
"""
|
|
from wiki_out.merge import is_human_verified, split_frontmatter
|
|
|
|
root = settings.wiki_dir.resolve()
|
|
if not doc_path.endswith(".md"):
|
|
doc_path += ".md"
|
|
target = (root / doc_path).resolve()
|
|
if not target.is_relative_to(root):
|
|
raise HTTPException(status_code=400, detail="잘못된 경로입니다 (wiki/ 루트 상대 경로만 허용)")
|
|
if not target.is_file():
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail=f"위키 문서 '{doc_path}' 이(가) 없습니다. "
|
|
f"경로는 wiki/ 루트 상대(예: programs/ZFIR10070.md)여야 합니다.",
|
|
)
|
|
text = target.read_text(encoding="utf-8")
|
|
fm, _body = split_frontmatter(text)
|
|
return {
|
|
"path": target.relative_to(root).as_posix(),
|
|
"human_verified": is_human_verified(fm), # 사람 교정 문서를 더 신뢰 (계획서 §6-3 규칙 0)
|
|
"content": text,
|
|
}
|
|
|
|
|
|
@app.get("/wiki-viewer", response_class=HTMLResponse)
|
|
def wiki_viewer() -> str:
|
|
"""위키 뷰어 — 요청 시점의 wiki/ 를 스캔해 렌더 (실시간)."""
|
|
from wiki_out.viewer import build_html
|
|
|
|
return build_html()
|
|
|
|
|
|
@app.get("/dashboard", response_class=HTMLResponse)
|
|
def dashboard() -> str:
|
|
"""관측소 대시보드 — 요청 시점의 index.db 상태를 담아 렌더 (실시간)."""
|
|
with _summarizing_lock:
|
|
running = sorted(_summarizing)
|
|
return render_dashboard(running)
|
|
|
|
|
|
@app.get("/summaries/status")
|
|
def summaries_status() -> dict:
|
|
"""요약 진행 현황과 LLM 사용량/비용 로그 (관측소 대시보드·gateway용)."""
|
|
con = connect()
|
|
try:
|
|
per_program = [dict(r) for r in con.execute(
|
|
"SELECT u.program, COUNT(*) AS eligible, "
|
|
"SUM(CASE WHEN u.summary_status='done' THEN 1 ELSE 0 END) AS done, "
|
|
"SUM(CASE WHEN u.summary_status='failed' THEN 1 ELSE 0 END) AS failed, "
|
|
"COALESCE(SUM(u.chunk_count),0) AS chunks, "
|
|
"MAX(p.summary_status) AS program_summary "
|
|
"FROM unit u JOIN program p ON p.name=u.program "
|
|
"WHERE u.unit_type IN ('FORM','METHOD','FUNCTION','MODULE','EVENT') "
|
|
"GROUP BY u.program ORDER BY u.program")]
|
|
totals = con.execute(
|
|
"SELECT COUNT(*) AS runs, COALESCE(SUM(calls),0) AS calls, "
|
|
"COALESCE(SUM(prompt_tokens),0) AS prompt_tokens, "
|
|
"COALESCE(SUM(completion_tokens),0) AS completion_tokens, "
|
|
"COALESCE(SUM(cost_usd),0) AS cost_usd FROM llm_usage_log").fetchone()
|
|
log = [dict(r) for r in con.execute(
|
|
"SELECT * FROM llm_usage_log ORDER BY id DESC LIMIT 50")]
|
|
failures = [dict(r) for r in con.execute(
|
|
"SELECT program, unit_id, include, unit_type, name, summary_error "
|
|
"FROM unit WHERE summary_status='failed' ORDER BY program, include, line_start")]
|
|
with _summarizing_lock:
|
|
running = sorted(_summarizing)
|
|
return {
|
|
"model": settings.llm_model,
|
|
"llm_enabled": settings.llm_enabled(),
|
|
"running": running,
|
|
"programs": per_program,
|
|
"failures": failures,
|
|
"usage_totals": dict(totals),
|
|
"usage_log": log,
|
|
"job_queue": _job_queue_status(),
|
|
}
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
def _job_queue_status() -> dict:
|
|
"""file 백엔드 작업 큐 현황 — LLM 키가 없을 때 진행 상황을 여기로 본다."""
|
|
from summarize.jobs import _index_rows, _status, jobs_dir
|
|
|
|
d = jobs_dir()
|
|
rows = _index_rows(d)
|
|
pending = [r for r in rows if _status(d, r) == "pending"]
|
|
return {
|
|
"dir": str(d),
|
|
"total": len(rows),
|
|
"pending": len(pending),
|
|
"answered": len(rows) - len(pending),
|
|
"next": [
|
|
{"job_id": r["job_id"], "task": r.get("task", ""), "program": r.get("program", ""),
|
|
"unit_id": r.get("unit_id", "")}
|
|
for r in pending[:20]
|
|
],
|
|
}
|
|
|
|
|
|
@app.get("/summaries/jobs")
|
|
def summaries_jobs() -> dict:
|
|
"""LLM 키 없이 돌리는 작업 큐 조회 (프롬프트 대기 목록).
|
|
|
|
채우는 방법은 `python -m summarize.jobs` 참고. API 로 응답을 받지는 않는다 —
|
|
프롬프트 본문이 커서 파일로 주고받는 쪽이 안전하다.
|
|
"""
|
|
return _job_queue_status()
|
|
|
|
|
|
# ===== LLM 설정 (관측소 대시보드에서 요약 모델 전환) =====
|
|
|
|
_MODEL_RE = re.compile(r"^[A-Za-z0-9._\-/:]+$")
|
|
ENV_PATH = ROOT / ".env" # 테스트에서 monkeypatch
|
|
|
|
|
|
def _persist_env(key: str, value: str) -> None:
|
|
"""ENV_PATH 의 key= 줄을 교체(없으면 추가) — 서버 재시작 후에도 유지되게."""
|
|
lines = ENV_PATH.read_text(encoding="utf-8").splitlines() if ENV_PATH.exists() else []
|
|
for i, ln in enumerate(lines):
|
|
if ln.startswith(f"{key}="):
|
|
lines[i] = f"{key}={value}"
|
|
break
|
|
else:
|
|
lines.append(f"{key}={value}")
|
|
ENV_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
|
|
|
|
@app.get("/settings/llm")
|
|
def get_llm_settings() -> dict:
|
|
return {
|
|
"model": settings.llm_model,
|
|
"enabled": settings.llm_enabled(),
|
|
}
|
|
|
|
|
|
@app.put("/settings/llm")
|
|
def put_llm_settings(payload: dict = Body()) -> dict:
|
|
model = str(payload.get("model", "")).strip()
|
|
if not model or not _MODEL_RE.match(model):
|
|
raise HTTPException(status_code=400, detail="model 형식이 잘못됐습니다 (영숫자 · . - / : 만 허용)")
|
|
settings.llm_model = model # 즉시 반영 — 다음 요약 배치부터 이 모델로 호출
|
|
_persist_env("LLM_MODEL", model)
|
|
return {"model": model}
|
|
|
|
|
|
def main() -> None:
|
|
import argparse
|
|
|
|
import uvicorn
|
|
|
|
ap = argparse.ArgumentParser(description="Stage 5 — 질의 API / 관측소 / 위키 뷰어")
|
|
ap.add_argument("--host", default=settings.index_host)
|
|
ap.add_argument("--port", type=int, default=settings.index_port)
|
|
ap.add_argument("--reload", action="store_true",
|
|
help="소스 변경 시 자동 재시작. 템플릿(html)은 요청마다 다시 읽지만 "
|
|
"파이썬 모듈은 프로세스에 캐시되므로, 코드를 고치는 중이면 이 옵션이 필요하다")
|
|
args = ap.parse_args()
|
|
if args.reload:
|
|
uvicorn.run("query.api:app", host=args.host, port=args.port, reload=True)
|
|
else:
|
|
uvicorn.run(app, host=args.host, port=args.port)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|