[AI] 파인튜닝 - google-adk 를 api_server 로 구동하자.

Instruction Dataset 을 쉽게 만드려다가 google-adk 를 다시 보게 되었는데 api server 처럼 사용이 가능하다고 하여 조금 정리해 봤다.

1. agent.py

import json
import os
from typing import List, Dict, Set, Tuple
from google.adk.agents import Agent

# ------------------ 프롬프트 생성 ------------------
def generate_qa_prompt(context: str, batch_index: int, batch_size: int = 10) -> str:
    return (
        f"[문장]: {context}\n\n"
        f"이 문장을 바탕으로 반드시 {batch_size}개의 서로 다른 질문과 그에 대한 답변을 생성해 주세요.\n"
        "각 질문과 답변은 중복되지 않아야 하며, 표현 방식도 다양해야 합니다.\n"
        "형식은 JSONL이며, 각 줄은 아래 형식과 완전히 일치해야 합니다:\n"
        '{"instruction": "질문", "input": "", "output": "답변"}\n\n'
        "예시:\n"
        '{"instruction": "서울은 어떤 나라의 수도인가요?", "input": "", "output": "대한민국의 수도는 서울입니다."}\n'
        '{"instruction": "사과에 많이 들어있는 영양소는?", "input": "", "output": "비타민 C와 식이섬유입니다."}\n\n'
        f"{batch_index * batch_size + 1}번부터 {batch_index * batch_size + batch_size}번까지 생성해 주세요.\n"
        "※ 반드시 정확히 10개를 생성해 주세요."
    )

# ------------------ 파싱 / 중복 제거 ------------------
def parse_jsonl(text: str) -> List[Dict[str, str]]:
    return [json.loads(line) for line in text.strip().splitlines() if line.strip()]

def normalize(text: str) -> str:
    import re
    return re.sub(r"\s+", " ", text.strip().lower())

def remove_duplicates(qa_list: List[Dict[str, str]], existing_set: Set[Tuple[str, str]]) -> List[Dict[str, str]]:
    seen = set(existing_set)
    unique = []
    for qa in qa_list:
        key = (normalize(qa['instruction']), normalize(qa['output']))
        if key not in seen:
            seen.add(key)
            unique.append(qa)
    return unique

# ------------------ 기존 결과 불러오기 ------------------
def load_existing_qa(path: str) -> Tuple[List[Dict[str, str]], Set[Tuple[str, str]]]:
    if not os.path.exists(path):
        return [], set()
    with open(path, "r", encoding="utf-8") as f:
        qa_list = [json.loads(line) for line in f if line.strip()]
    key_set = {(normalize(q["instruction"]), normalize(q["output"])) for q in qa_list}
    return qa_list, key_set

# ------------------ 생성 에이전트 ------------------
root_agent = Agent(
    name="root_agent",
    model="gemini-2.0-flash",
    description="Generates batches of Q&A pairs based on a sentence.",
    instruction="Generate non-redundant Q&A pairs in JSONL format from a sentence. Avoid duplication in questions and answers.",
)

2. agent 를 api_server 로 실행

% adk api_server

3. 사용자와 세션 설정

• URL
  http://0.0.0.0:8000/apps/multi-agent/users/exoluse/sessions/session_exoluse

• Method
  POST

• Header
  Content-Type : application/json

• Body

{
    "state": {"key1": "val1", "key2": 42}
}

• 결과
  user : exoluse
  session : session_exoluse
  app_name : multi-agent

4. App 실행

• URL
  http://0.0.0.0:8000/run_sse

• Method
  POST

• Header
  Content-Type : application/json

• Body

{
"app_name": "multi-agent",
"user_id": "exoluse",
"session_id": "session_exoluse",
"new_message": {
    "role": "user",
    "parts": [{
    "text": "나는 exoluse 입니다. 나는 한국에 살고 있습니다."
    }]
}
}

• 결과

{
    "content": {
        "parts": [
            {
                "text": "```jsonl\n{\"question\": \"나는 누구입니까?\", \"answer\": \"당신은 exoluse입니다.\"}\n{\"question\": \"어디에 살고 있습니까?\", \"answer\": \"한국에 살고 있습니다.\"}\n{\"question\": \"어느 나라에 살고 있습니까?\", \"answer\": \"한국\"}\n```\n"
            }
        ],
        "role": "model"
    },
    "invocation_id": "e-29b034c4-7248-456d-ae4e-a1df612aabbb",
    "author": "root_agent",
    "actions": {
        "state_delta": {},
        "artifact_delta": {},
        "requested_auth_configs": {}
    },
    "id": "7rw4dTxI",
    "timestamp": 1747145235.559092
}

5. 서버는 돌고있음

6. 생각보다 토큰 적게 씀