읽기 전용 도구 9종으로 카탈로그에 묻고 답하다 — Function Calling 설계

TOOLS: dict[str, dict] = {
    "search_datasets": {
        "fn": search_datasets,
        "schema": {
            "type": "function",
            "function": {
                "name": "search_datasets",
                "description": "데이터 카탈로그를 시맨틱 검색한다. 사용자가 "
                               "어떤 데이터/테이블을 찾거나 언급하면 먼저 이 도구로 찾는다.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {"type": "string", "description": "검색어 (한국어 가능)"},
                        "limit": {"type": "integer", "description": "결과 수 (기본 5)"},
                    },
                    "required": ["query"],
                },
            },
        },
    },
    ...
}

MAX_TOOL_ROUNDS = 6   # 도구 호출 무한 루프 방지 — 일반 질문은 2~3회면 충분
 
for round_no in range(MAX_TOOL_ROUNDS):
    # 1) 스트리밍 호출 — 답변 토큰은 즉시 흘리고, tool_calls 는 조립
    for ev in llm.chat_stream(messages, tools=tools):
        if ev["type"] == "content":
            yield {"type": "text_delta", "data": {"text": ev["text"]}}
        elif ev["type"] == "final":
            msg, usage = ev["message"], ev["usage"]
 
    tool_calls = (msg or {}).get("tool_calls") or []
    if not tool_calls:
        # 2) 도구 호출 없음 = 최종 답변 → 종료
        yield {"type": "done", "data": {}}
        return
 
    # 3) 도구 실행 라운드 — 결과를 role=tool 메시지로 붙여 다시 호출
    messages.append(msg)
    for tc in tool_calls:
        name = tc["function"]["name"]
        args = json.loads(tc["function"]["arguments"] or "{}")
        yield {"type": "tool_call", "data": {"id": call_id, "name": name, "args": args}}
        tool_result = run_tool(tool_ctx, name, args)
        yield {"type": "tool_result", "data": {"id": call_id, "name": name, "result": tool_result}}
        messages.append({"role": "tool", "tool_call_id": call_id,
                         "content": json.dumps(tool_result, ensure_ascii=False)})

class ToolContext:
    """도구 실행 컨텍스트 — 카탈로그 API 주소와 사용자 토큰."""
 
    def __init__(self, api_url: str, user_token: str) -> None:
        self.base = api_url.rstrip("/") + "/api/v1"
        self.token = user_token
 
    def get(self, path: str):
        """사용자 토큰으로 GET."""
        req = urllib.request.Request(
            self.base + path,
            headers={"Authorization": f"Bearer {self.token}"})
        with urllib.request.urlopen(req, timeout=30) as resp:
            return json.loads(resp.read())

def get_quality(ctx: ToolContext, dataset_id: int) -> dict:
    """품질 상태 — 점수·실패 규칙·위반 샘플."""
    ...
    failed = [
        {"rule": r.get("rule_name"), "type": r.get("check_type"),
         "column": r.get("column_name"), "severity": r.get("severity"),
         "detail": r.get("detail"),
         "samples": (r.get("failed_samples") or [])[:2] or None}  # 위반 샘플 2행만
        for r in results if r.get("passed") == "false"             # 실패 규칙만
    ]

_FORBIDDEN_SQL = re.compile(
    r"\b(insert|update|delete|drop|alter|create|truncate|"
    r"grant|revoke|merge|call|exec|execute)\b", re.IGNORECASE)
 
def validate_sql(ctx, sql: str) -> dict:
    stripped = re.sub(r"--[^\n]*", "", sql)              # 주석 제거
    stripped = stripped.strip().rstrip(";").strip()
    if ";" in stripped:
        return {"valid": False, "reason": "다중 문장(;)은 허용되지 않습니다"}
    head = stripped.split(None, 1)[0].lower()
    if head not in ("select", "with"):
        return {"valid": False, "reason": "SELECT/WITH 조회 쿼리만 허용됩니다"}
    if _FORBIDDEN_SQL.search(stripped):
        return {"valid": False, "reason": "허용되지 않는 키워드"}
    return {"valid": True, "reason": "통과"}

def run_tool(ctx, name: str, args: dict) -> dict:
    """모든 예외를 결과(오류)로 변환해 루프가 끊기지 않게 한다."""
    tool = TOOLS.get(name)
    if not tool:
        return {"error": f"알 수 없는 도구: {name}"}
    try:
        return tool["fn"](ctx, **args)
    except Exception as e:
        return {"error": f"도구 실행 실패: {e}"}