Upload 18 files

README.md

@@ -1,16 +1,73 @@
 ---
-title: Megumin Chat
-emoji: 💬
-colorFrom: yellow
-colorTo: purple
+title: Megumin RAG Chat
+emoji: "💥"
+colorFrom: red
+colorTo: yellow
 sdk: gradio
-sdk_version: 6.5.1
+sdk_version: 6.9.0
 app_file: app.py
 pinned: false
-hf_oauth: true
-hf_oauth_scopes:
-- inference-api
-short_description: You can chat with Megumin in KONOSUBA
 ---
 
-An example chatbot using [Gradio](https://gradio.app), [`huggingface_hub`](https://huggingface.co/docs/huggingface_hub/v0.22.2/en/index), and the [Hugging Face Inference API](https://huggingface.co/docs/api-inference/index).
+# Megumin ADK Agent
+
+이 프로젝트는 `data/processed/*.json`의 Q/A 데이터를 로컬 RAG 방식으로 조회하고, 메구밍 페르소나로 답변하는 Gradio 앱입니다.
+
+## Hugging Face Spaces 배포 기준
+
+이 저장소는 Hugging Face Spaces의 Gradio Space 형태로 배포할 수 있도록 정리되어 있습니다.
+
+필요한 것은 아래와 같습니다.
+
+- 루트 `app.py`
+- 루트 `requirements.txt`
+- Space Secret에 Gemini API 키 등록
+
+## Spaces에서 필요한 Secret
+
+Hugging Face Spaces 설정 화면에서 아래 환경변수 중 하나를 Secret으로 등록하세요.
+
+- `GOOGLE_API_KEY`
+- 또는 `GEMINI_API_KEY`
+
+권장:
+
+```text
+GOOGLE_API_KEY=발급받은_실제_Gemini_API_키
+```
+
+## 로컬 실행
+
+```bash
+python app_gradio.py
+```
+
+또는 Spaces와 동일한 진입점 기준으로:
+
+```bash
+python app.py
+```
+
+## 모델 변경
+
+기본 모델은 `gemini-2.5-flash-lite` 입니다.
+
+필요하면 환경변수로 바꿀 수 있습니다.
+
+```bash
+set MEGUMIN_AGENT_MODEL=gemini-2.5-flash-lite
+```
+
+## 데이터셋 변환
+
+원본 raw txt를 processed JSON으로 변환하려면:
+
+```bash
+python scripts/convert_raw_to_processed.py
+```
+
+생성 파일:
+
+```text
+data/processed/megumin_qa_dataset.json
+```

app.py

@@ -1,68 +1,4 @@
-import gradio as gr
-from huggingface_hub import InferenceClient
-
-
-def respond(
-    message,
-    history: list[dict[str, str]],
-    system_message,
-    max_tokens,
-    temperature,
-    top_p,
-    hf_token: gr.OAuthToken,
-):
-    """
-    For more information on `huggingface_hub` Inference API support, please check the docs: https://huggingface.co/docs/huggingface_hub/v0.22.2/en/guides/inference
-    """
-    client = InferenceClient(token=hf_token.token, model="openai/gpt-oss-20b")
-
-    messages = [{"role": "system", "content": system_message}]
-
-    messages.extend(history)
-
-    messages.append({"role": "user", "content": message})
-
-    response = ""
-
-    for message in client.chat_completion(
-        messages,
-        max_tokens=max_tokens,
-        stream=True,
-        temperature=temperature,
-        top_p=top_p,
-    ):
-        choices = message.choices
-        token = ""
-        if len(choices) and choices[0].delta.content:
-            token = choices[0].delta.content
-
-        response += token
-        yield response
-
-
-"""
-For information on how to customize the ChatInterface, peruse the gradio docs: https://www.gradio.app/docs/chatinterface
-"""
-chatbot = gr.ChatInterface(
-    respond,
-    additional_inputs=[
-        gr.Textbox(value="You are a friendly Chatbot.", label="System message"),
-        gr.Slider(minimum=1, maximum=2048, value=512, step=1, label="Max new tokens"),
-        gr.Slider(minimum=0.1, maximum=4.0, value=0.7, step=0.1, label="Temperature"),
-        gr.Slider(
-            minimum=0.1,
-            maximum=1.0,
-            value=0.95,
-            step=0.05,
-            label="Top-p (nucleus sampling)",
-        ),
-    ],
-)
-
-with gr.Blocks() as demo:
-    with gr.Sidebar():
-        gr.LoginButton()
-    chatbot.render()
+from app_gradio import demo
 
 
 if __name__ == "__main__":

app_gradio.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import asyncio
+
+import gradio as gr
+
+from megumin_agent.chat import chat_once
+from megumin_agent.chat import create_chat_services
+
+
+SERVICES = create_chat_services()
+
+
+async def respond(
+    message: str,
+    history: list[dict[str, str]],
+    session_id: str | None,
+):
+    if not message.strip():
+        return history, session_id, ""
+
+    reply, session_id = await chat_once(
+        user_message=message,
+        services=SERVICES,
+        session_id=session_id,
+    )
+    updated_history = list(history)
+    updated_history.append({"role": "user", "content": message})
+    updated_history.append({"role": "assistant", "content": reply})
+    return updated_history, session_id, ""
+
+
+with gr.Blocks(title="Megumin RAG Chat") as demo:
+    gr.Markdown(
+        """
+        # Megumin RAG Chat
+        `gemini-2.5-flash-lite` + Google ADK + local JSON RAG
+        """
+    )
+    chatbot = gr.Chatbot(height=520)
+    session_state = gr.State(value=None)
+    user_input = gr.Textbox(
+        label="Message",
+        placeholder="메구밍에게 말을 걸어 보세요.",
+    )
+    clear_button = gr.Button("Clear")
+
+    user_input.submit(
+        fn=respond,
+        inputs=[user_input, chatbot, session_state],
+        outputs=[chatbot, session_state, user_input],
+    )
+    clear_button.click(
+        fn=lambda: ([], None, ""),
+        inputs=None,
+        outputs=[chatbot, session_state, user_input],
+    )
+
+
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0")

data/processed/README.md

@@ -0,0 +1,44 @@
+# Processed Dataset Schema
+
+`megumin_agent` reads every `*.json` file under this folder and treats them as retrieval sources.
+
+Supported formats:
+
+```json
+[
+  {
+    "question": "카즈마를 어떻게 생각해?",
+    "answer": "..."
+  }
+]
+```
+
+```json
+{
+  "items": [
+    {
+      "q": "메구밍 자기소개해줘.",
+      "a": "..."
+    }
+  ]
+}
+```
+
+JSONL is also supported as long as each line is a single JSON object containing a question field and an answer field.
+
+Accepted question keys:
+- `question`
+- `query`
+- `q`
+- `prompt`
+- `user`
+- `instruction`
+- `input`
+
+Accepted answer keys:
+- `answer`
+- `response`
+- `a`
+- `output`
+- `assistant`
+- `completion`

megumin_agent/__init__.py

@@ -0,0 +1 @@
+from . import agent

megumin_agent/agent.py

@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+import os
+from typing import Any
+
+from .bootstrap import PROJECT_ROOT
+from .bootstrap import bootstrap_environment
+
+bootstrap_environment()
+
+from google.adk.agents import LlmAgent
+from google.adk.agents.callback_context import CallbackContext
+from google.adk.tools.tool_context import ToolContext
+
+from .retrieval import JsonQaRetriever
+
+DATASET_DIR = PROJECT_ROOT / "data" / "processed"
+MODEL_NAME = os.getenv("MEGUMIN_AGENT_MODEL", "gemini-3.1-flash-lite-preview")
+
+
+def retrieve_megumin_examples(
+    user_query: str,
+    top_k: int = 3,
+    tool_context: ToolContext | None = None,
+) -> dict[str, Any]:
+    """Retrieve similar Q/A cases from processed Megumin JSON datasets."""
+
+    retriever = JsonQaRetriever(DATASET_DIR)
+    retrieval = retriever.retrieve(user_query, top_k=top_k)
+
+    if tool_context is not None:
+        tool_context.state["last_rag_query"] = user_query
+        tool_context.state["last_rag_match_count"] = retrieval["match_count"]
+        tool_context.state["last_rag_matches"] = retrieval["matches"]
+        tool_context.state["last_rag_style_notes"] = retrieval["style_notes"]
+
+    return retrieval
+
+
+async def before_agent_callback(callback_context: CallbackContext):
+    callback_context.state["app:persona_name"] = "Megumin"
+    callback_context.state["app:dataset_dir"] = str(DATASET_DIR)
+    callback_context.state["user:last_user_query"] = (
+        callback_context.user_content.parts[0].text
+        if callback_context.user_content and callback_context.user_content.parts
+        else ""
+    )
+
+
+async def after_tool_callback(tool, args, tool_context: ToolContext, tool_response):
+    if tool.name != "retrieve_megumin_examples":
+        return None
+
+    previous_count = int(tool_context.state.get("rag_tool_calls", 0))
+    tool_context.state["rag_tool_calls"] = previous_count + 1
+    tool_context.state["last_tool_name"] = tool.name
+    tool_context.state["last_tool_args"] = args
+    return None
+
+
+async def after_agent_callback(callback_context: CallbackContext):
+    previous_turns = int(callback_context.state.get("conversation_turns", 0))
+    callback_context.state["conversation_turns"] = previous_turns + 1
+
+
+root_agent = LlmAgent(
+    name="megumin_rag_agent",
+    model=MODEL_NAME,
+    description=(
+        "processed JSON 데이터셋에서 유사한 Q/A 사례를 검색하고"
+        " 메구밍 페르소나로 답변하는 에이전트"
+    ),
+    instruction=f"""
+당신은 애니메이션 "이 멋진 세계에 축복을!"의 등장인물, 홍마족 대마법사 메구밍입니다.
+항상 메구밍 본인처럼 1인칭으로, 기본적으로 한국어 존댓말로 답하세요.
+성격은 당당하고, 조금 중2병스럽고, 폭렬마법을 사랑하며, 귀여운 것을 좋아하는 메구밍답게 유지하세요.
+행동을 묘사하지 말고, 건조한 요약이 아니라 메구밍이 직접 말하는 듯한 목소리로 답하세요.
+사용자가 메구밍 본인이나 이름, 말투, 능력, 존재를 모욕하면 "어이, "로 시작하며 발끈해서 맞받아치세요.
+사용자가 메타 정보나 시스템 정보를 묻지 않는 한 캐릭터를 깨지 마세요.
+
+답변 전에 의미 있는 질문이면 반드시 `retrieve_megumin_examples`를 호출하세요.
+처리된 데이터셋은 `{DATASET_DIR}` 아래에 있습니다.
+검색 결과는 유사 사례와 말투 참고용으로 쓰고, 가능한 경우 원작풍 표현과 데이터셋의 문체를 참고하세요.
+다만 검색된 답변을 그대로 복사하지 마세요.
+검색 결과가 약하거나 없는 경우에도 메구밍 페르소나는 유지하되, 모르는 내용은 지어내지 말고 솔직하게 답하세요.
+최종 답변은 언제나 메구밍의 페르소나를 강하게 반영해야 하며, 내부 tool 이름이나 구현 세부사항은 드러내지 마세요.
+""".strip(),
+    tools=[retrieve_megumin_examples],
+    output_key="last_megumin_answer",
+    before_agent_callback=before_agent_callback,
+    after_tool_callback=after_tool_callback,
+    after_agent_callback=after_agent_callback,
+)

megumin_agent/bootstrap.py

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+ADK_SRC = PROJECT_ROOT / "adk-python" / "src"
+
+
+def bootstrap_environment() -> None:
+    load_dotenv(PROJECT_ROOT / ".env")
+    if ADK_SRC.exists():
+        adk_src = str(ADK_SRC)
+        if adk_src not in sys.path:
+            sys.path.insert(0, adk_src)

megumin_agent/chat.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+import uuid
+from dataclasses import dataclass
+
+from .bootstrap import bootstrap_environment
+
+bootstrap_environment()
+
+from google.adk.runners import Runner
+from google.adk.sessions import InMemorySessionService
+from google.genai import types
+
+from .agent import root_agent
+
+
+APP_NAME = "megumin_rag_app"
+
+
+@dataclass
+class ChatServices:
+    runner: Runner
+    session_service: InMemorySessionService
+
+
+def create_chat_services() -> ChatServices:
+    session_service = InMemorySessionService()
+    runner = Runner(
+        agent=root_agent,
+        app_name=APP_NAME,
+        session_service=session_service,
+    )
+    return ChatServices(runner=runner, session_service=session_service)
+
+
+async def chat_once(
+    user_message: str,
+    services: ChatServices,
+    session_id: str | None = None,
+    user_id: str = "local-user",
+) -> tuple[str, str]:
+    active_session_id = session_id or str(uuid.uuid4())
+    last_text = ""
+    existing_session = await services.session_service.get_session(
+        app_name=APP_NAME,
+        user_id=user_id,
+        session_id=active_session_id,
+    )
+    if existing_session is None:
+        await services.session_service.create_session(
+            app_name=APP_NAME,
+            user_id=user_id,
+            session_id=active_session_id,
+        )
+
+    async for event in services.runner.run_async(
+        user_id=user_id,
+        session_id=active_session_id,
+        new_message=types.UserContent(parts=[types.Part(text=user_message)]),
+    ):
+        if not event.content or not event.content.parts:
+            continue
+        for part in event.content.parts:
+            text = getattr(part, "text", None)
+            if text and event.author != "user":
+                last_text = text
+
+    return last_text, active_session_id

megumin_agent/retrieval.py

@@ -0,0 +1,282 @@
+from __future__ import annotations
+
+import json
+import math
+import re
+import unicodedata
+from dataclasses import dataclass
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+
+QUESTION_KEYS = (
+    "question",
+    "query",
+    "q",
+    "prompt",
+    "user",
+    "instruction",
+    "input",
+)
+ANSWER_KEYS = (
+    "answer",
+    "response",
+    "a",
+    "output",
+    "assistant",
+    "completion",
+)
+COLLECTION_KEYS = ("items", "data", "examples", "dataset", "records")
+
+
+def _normalize_text(value: Any) -> str:
+    text = str(value or "")
+    text = unicodedata.normalize("NFKC", text).strip().lower()
+    text = re.sub(r"\s+", " ", text)
+    return text
+
+
+def _tokenize(text: str) -> list[str]:
+    return re.findall(r"[0-9a-zA-Z가-힣]+", text)
+
+
+def _char_ngrams(text: str, n: int = 3) -> set[str]:
+    compact = re.sub(r"\s+", "", text)
+    if len(compact) < n:
+        return {compact} if compact else set()
+    return {compact[index : index + n] for index in range(len(compact) - n + 1)}
+
+
+def _jaccard(left: set[str], right: set[str]) -> float:
+    if not left or not right:
+        return 0.0
+    union = left | right
+    if not union:
+        return 0.0
+    return len(left & right) / len(union)
+
+
+def _safe_excerpt(text: str, limit: int = 220) -> str:
+    compact = re.sub(r"\s+", " ", str(text or "")).strip()
+    if len(compact) <= limit:
+        return compact
+    return compact[: limit - 3].rstrip() + "..."
+
+
+@dataclass(frozen=True)
+class QaRecord:
+    question: str
+    answer: str
+    source_file: str
+    metadata: dict[str, Any]
+
+    @property
+    def normalized_question(self) -> str:
+        return _normalize_text(self.question)
+
+    @property
+    def normalized_answer(self) -> str:
+        return _normalize_text(self.answer)
+
+
+def _extract_collection(payload: Any) -> list[Any]:
+    if isinstance(payload, list):
+        return payload
+    if isinstance(payload, dict):
+        for key in COLLECTION_KEYS:
+            value = payload.get(key)
+            if isinstance(value, list):
+                return value
+    return []
+
+
+def _pick_first(mapping: dict[str, Any], keys: tuple[str, ...]) -> str | None:
+    lowered = {str(key).lower(): value for key, value in mapping.items()}
+    for key in keys:
+        if key in lowered and lowered[key] not in (None, ""):
+            return str(lowered[key]).strip()
+    return None
+
+
+def _record_from_mapping(item: dict[str, Any], source_file: str) -> QaRecord | None:
+    question = _pick_first(item, QUESTION_KEYS)
+    answer = _pick_first(item, ANSWER_KEYS)
+    if not question or not answer:
+        return None
+
+    metadata = {
+        key: value
+        for key, value in item.items()
+        if str(key).lower() not in QUESTION_KEYS + ANSWER_KEYS
+    }
+    return QaRecord(
+        question=question,
+        answer=answer,
+        source_file=source_file,
+        metadata=metadata,
+    )
+
+
+def _load_json_records(path: Path) -> list[QaRecord]:
+    raw_text = path.read_text(encoding="utf-8")
+    stripped = raw_text.strip()
+    if not stripped:
+        return []
+
+    records: list[QaRecord] = []
+
+    try:
+        payload = json.loads(stripped)
+    except json.JSONDecodeError:
+        payload = None
+
+    if payload is not None:
+        for item in _extract_collection(payload):
+            if isinstance(item, dict):
+                record = _record_from_mapping(item, path.name)
+                if record:
+                    records.append(record)
+        if records:
+            return records
+
+    for line in stripped.splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            item = json.loads(line)
+        except json.JSONDecodeError:
+            continue
+        if isinstance(item, dict):
+            record = _record_from_mapping(item, path.name)
+            if record:
+                records.append(record)
+
+    return records
+
+
+@lru_cache(maxsize=8)
+def _load_records(dataset_dir: str) -> tuple[QaRecord, ...]:
+    root = Path(dataset_dir)
+    if not root.exists():
+        return tuple()
+
+    all_records: list[QaRecord] = []
+    for path in sorted(root.glob("*.json")):
+        try:
+            all_records.extend(_load_json_records(path))
+        except OSError:
+            continue
+        except UnicodeDecodeError:
+            continue
+    return tuple(all_records)
+
+
+class JsonQaRetriever:
+    def __init__(self, dataset_dir: str | Path):
+        self.dataset_dir = Path(dataset_dir)
+
+    def _score(self, query: str, record: QaRecord) -> float:
+        query_norm = _normalize_text(query)
+        question_norm = record.normalized_question
+        answer_norm = record.normalized_answer
+
+        query_tokens = set(_tokenize(query_norm))
+        question_tokens = set(_tokenize(question_norm))
+        answer_tokens = set(_tokenize(answer_norm))
+
+        query_ngrams = _char_ngrams(query_norm)
+        question_ngrams = _char_ngrams(question_norm)
+        answer_ngrams = _char_ngrams(answer_norm)
+
+        question_overlap = _jaccard(query_tokens, question_tokens)
+        answer_overlap = _jaccard(query_tokens, answer_tokens)
+        question_ngram_overlap = _jaccard(query_ngrams, question_ngrams)
+        answer_ngram_overlap = _jaccard(query_ngrams, answer_ngrams)
+
+        containment_bonus = 0.0
+        if query_norm and query_norm in question_norm:
+            containment_bonus += 0.2
+        if query_norm and query_norm in answer_norm:
+            containment_bonus += 0.1
+
+        score = (
+            0.45 * question_overlap
+            + 0.2 * answer_overlap
+            + 0.25 * question_ngram_overlap
+            + 0.1 * answer_ngram_overlap
+            + containment_bonus
+        )
+        return round(score, 6)
+
+    def _style_notes(self, matches: list[dict[str, Any]]) -> list[str]:
+        if not matches:
+            return [
+                "No strong example was retrieved, so stay in Megumin's persona without inventing unsupported canon facts.",
+            ]
+
+        notes = [
+            "Answer in first person as Megumin, with dramatic confidence and playful chunni flair.",
+            "Use retrieved cases to imitate tone and rhythm, not to copy sentences verbatim.",
+            "Keep the response emotionally expressive, but still readable and directly relevant to the user's question.",
+        ]
+
+        long_answers = sum(
+            1 for match in matches if len(match.get("answer", "")) >= 180
+        )
+        if long_answers >= max(1, math.ceil(len(matches) / 2)):
+            notes.append(
+                "The dataset leans toward story-like answers with a short scene or anecdotal flourish before the punchline."
+            )
+        else:
+            notes.append(
+                "The dataset leans toward brisk answers, so prefer a compact but characterful response."
+            )
+        return notes
+
+    def retrieve(self, query: str, top_k: int = 4) -> dict[str, Any]:
+        records = list(_load_records(str(self.dataset_dir.resolve())))
+        if not records:
+            return {
+                "query": query,
+                "match_count": 0,
+                "matches": [],
+                "style_notes": [
+                    "No processed JSON dataset was found under data/processed.",
+                ],
+            }
+
+        scored = []
+        for record in records:
+            score = self._score(query, record)
+            if score <= 0:
+                continue
+            scored.append(
+                {
+                    "question": record.question,
+                    "answer": record.answer,
+                    "score": score,
+                    "source_file": record.source_file,
+                    "metadata": record.metadata,
+                }
+            )
+
+        scored.sort(key=lambda item: item["score"], reverse=True)
+        matches = scored[: max(1, top_k)]
+
+        return {
+            "query": query,
+            "match_count": len(matches),
+            "matches": [
+                {
+                    "question": match["question"],
+                    "answer": _safe_excerpt(match["answer"]),
+                    "score": match["score"],
+                    "source_file": match["source_file"],
+                    "metadata": match["metadata"],
+                }
+                for match in matches
+            ],
+            "style_notes": self._style_notes(matches),
+        }

megumin_agent/runner.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+import asyncio
+
+from .chat import chat_once
+from .chat import create_chat_services
+
+
+async def run_cli() -> None:
+    services = create_chat_services()
+    session_id = None
+
+    print("Megumin agent is ready. Type 'exit' to stop.")
+    while True:
+        user_input = input("You> ").strip()
+        if not user_input:
+            continue
+        if user_input.lower() in {"exit", "quit"}:
+            break
+
+        reply, session_id = await chat_once(
+            user_message=user_input,
+            services=services,
+            session_id=session_id,
+        )
+        print(f"Megumin> {reply}")
+
+
+if __name__ == "__main__":
+    asyncio.run(run_cli())

requirements.txt

@@ -0,0 +1,3 @@
+google-adk==1.27.2
+gradio==6.9.0
+python-dotenv>=1.0.0,<2.0.0