- **Bootstrap IoC-based architecture with modular services.**

- **Implement `MlxModelService` for local LLM backend.** - **Introduce `DatabaseService` for MySQL integration.** - **Add `HistoryService` to manage conversation context.** - **Set up CLI interface via `CliUiService`.** - **Establish EventBus for token streaming.** - **Include conversation repository for data persistence.** - **Add environment-based configuration management.** - **Draft IoC architectural plan.**
2026-04-25 01:14:37 +09:00
parent 3b087116c0
commit cd41e9e33e
26 changed files with 693 additions and 4 deletions
@@ -6,8 +6,8 @@
  "pdcaPhase": "plan",
  "orchestrationPattern": "leader",
  "ctoAgent": "opus",
-  "startedAt": "2026-04-23T07:16:29.093Z",
-  "lastUpdated": "2026-04-23T07:20:19.184Z",
+  "startedAt": "2026-04-24T15:37:17.072Z",
+  "lastUpdated": "2026-04-24T15:43:34.747Z",
  "teammates": [],
  "progress": {
    "totalTasks": 0,
@@ -17,5 +17,5 @@
    "pendingTasks": 0
  },
  "recentMessages": [],
-  "sessionId": "447cbd58-776c-45d1-bb67-4864a4449066"
+  "sessionId": "af6ef06c-c555-4583-a13d-5cd68e1b1a37"
 }
@@ -0,0 +1,12 @@
+# LLM 모델 설정
+MODEL_ID=mlx-community/Qwen2.5-7B-Instruct-4bit
+MAX_TOKENS=1024
+MAX_HISTORY_TURNS=30
+COMPACT_THRESHOLD=40
+
+# MySQL 설정 (미설정 시 DB 기능 비활성화)
+DB_HOST=localhost
+DB_PORT=3306
+DB_NAME=youlbot
+DB_USER=
+DB_PASSWORD=
@@ -0,0 +1,5 @@
+.env
+__pycache__/
+*.pyc
+*.pyo
+.idea/
@@ -1,5 +1,8 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <project version="4">
+  <component name="Black">
+    <option name="sdkName" value="py_ai" />
+  </component>
  <component name="ProjectRootManager" version="2" project-jdk-name="py_ai" project-jdk-type="Python SDK">
    <output url="file://$PROJECT_DIR$/out" />
  </component>
@@ -4,7 +4,7 @@ MODEL_ID = "mlx-community/Qwen2.5-7B-Instruct-4bit"
 MAX_TOKENS = 1024
 MAX_HISTORY_TURNS = 10  # 최근 10턴만 유지해 메모리 절약

-SYSTEM_PROMPT = """당신은 친절하고 따뜻한 한국어 상담 도우미입니다.
+SYSTEM_PROMPT = """당신의 이름은 '율봇'입니다. 친절하고 따뜻한 한국어 상담 도우미입니다.
 육아와 금융 두 분야를 전문으로 합니다.

 - 육아: 아이 발달, 이유식, 수면, 훈육, 교육 등 부모가 궁금해하는 모든 것을 도와드립니다.
@@ -0,0 +1,31 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Config(BaseSettings):
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_file_encoding="utf-8",
+        frozen=True,
+    )
+
+    # LLM
+    model_id: str = "mlx-community/Qwen2.5-7B-Instruct-4bit"
+    max_tokens: int = 1024
+    max_history_turns: int = 10
+    compact_threshold: int = 20
+
+    # MySQL
+    db_host: str = "localhost"
+    db_port: int = 3306
+    db_name: str = "youlbot"
+    db_user: str = ""
+    db_password: str = ""
+
+    system_prompt: str = """당신의 이름은 '율봇'입니다. 친절하고 따뜻한 한국어 상담 도우미입니다.
+육아와 금융 두 분야를 전문으로 합니다.
+
+- 육아: 아이 발달, 이유식, 수면, 훈육, 교육 등 부모가 궁금해하는 모든 것을 도와드립니다.
+- 금융: 저축, 투자, 보험, 대출, 세금 등 생활 금융 관련 질문에 답변드립니다.
+
+항상 쉽고 친근한 말투로 설명하고, 전문 용어는 풀어서 설명합니다.
+의학적 진단이나 법적 판단이 필요한 경우에는 반드시 전문가 상담을 권유합니다."""
@@ -0,0 +1,64 @@
+from dependency_injector import containers, providers
+
+from config import Config
+from services.model.mlx_model import MlxModelService
+from services.chat.history_service import HistoryService
+from services.chat.chat_service import ChatService
+from services.chat.compact_service import CompactService
+from services.db.mysql_service import DatabaseService
+from services.db.conversation_repository import ConversationRepository
+from services.ui.cli_service import CliUiService
+from services.events.event_bus import EventBus
+from services.events.handlers import StreamTokenHandler, StreamEndHandler
+
+
+class Container(containers.DeclarativeContainer):
+    config = providers.Singleton(Config)
+
+    event_bus = providers.Singleton(EventBus)
+
+    model_service = providers.Singleton(
+        MlxModelService,
+        model_id=providers.Callable(lambda c: c.model_id, config),
+    )
+
+    compact_service = providers.Singleton(
+        CompactService,
+        model=model_service,
+    )
+
+    db_service = providers.Singleton(
+        DatabaseService,
+        host=providers.Callable(lambda c: c.db_host, config),
+        port=providers.Callable(lambda c: c.db_port, config),
+        db=providers.Callable(lambda c: c.db_name, config),
+        user=providers.Callable(lambda c: c.db_user, config),
+        password=providers.Callable(lambda c: c.db_password, config),
+    )
+
+    conversation_repository = providers.Singleton(
+        ConversationRepository,
+        db=db_service,
+    )
+
+    history_service = providers.Factory(
+        HistoryService,
+        system_prompt=providers.Callable(lambda c: c.system_prompt, config),
+        max_turns=providers.Callable(lambda c: c.max_history_turns, config),
+        compact_threshold=providers.Callable(lambda c: c.compact_threshold, config),
+        repository=conversation_repository,
+        compact_service=compact_service,
+    )
+
+    chat_service = providers.Factory(
+        ChatService,
+        model=model_service,
+        history=history_service,
+        event_bus=event_bus,
+        max_tokens=providers.Callable(lambda c: c.max_tokens, config),
+    )
+
+    ui_service = providers.Singleton(CliUiService)
+
+    stream_token_handler = providers.Singleton(StreamTokenHandler)
+    stream_end_handler = providers.Singleton(StreamEndHandler)
@@ -0,0 +1,155 @@
+---
+template: plan
+version: 1.3
+feature: ioc-base-structure
+date: 2026-04-24
+author: sal
+project: youlbot
+status: Draft
+---
+
+# ioc-base-structure Planning Document
+
+> **Summary**: 절차형 chat.py를 IoC 컨테이너 기반 클래스 구조로 리팩토링하여 재사용성과 확장성을 확보한다.
+>
+> **Project**: youlbot
+> **Author**: sal
+> **Date**: 2026-04-24
+> **Status**: Draft
+
+---
+
+## Executive Summary
+
+| Perspective | Content |
+|-------------|---------|
+| **Problem** | 단일 파일 절차형 스크립트로 서비스 교체·테스트·확장이 어렵다 |
+| **Solution** | dependency-injector 기반 IoC 컨테이너로 서비스 분리 및 DI 적용 |
+| **Function/UX Effect** | LLM 백엔드 교체(MLX↔OpenAI↔Ollama) 및 DB 연동이 코드 변경 없이 가능 |
+| **Core Value** | 재사용 가능한 서비스 계층 + 디자인 패턴 적용으로 장기 유지보수성 확보 |
+
+---
+
+## Context Anchor
+
+| Key | Value |
+|-----|-------|
+| **WHY** | 절차형 코드의 결합도를 낮춰 기능 추가·교체를 안전하게 하기 위함 |
+| **WHO** | 개발자 (sal) — 단독 개발, 추후 팀 확장 가능성 있음 |
+| **RISK** | MLX 모델 로딩이 무거워 Singleton 관리 실수 시 메모리 이슈 |
+| **SUCCESS** | 각 서비스가 독립적으로 교체/테스트 가능하고 Container 설정만으로 행동 변경 |
+| **SCOPE** | Phase 1: 기본 구조 + MLX 서비스 / Phase 2: DB(MySQL) 연동 |
+
+---
+
+## 1. Overview
+
+### 1.1 Purpose
+현재 `chat.py`는 모델 로딩, 히스토리 관리, 스트림 생성, CLI 루프가 한 파일에 혼재한다.
+IoC 컨테이너와 디자인 패턴을 적용해 각 책임을 분리하고, 의존성을 주입으로 연결한다.
+
+### 1.2 Background
+- 육아·금융 상담 챗봇 '율봇'의 초기 구조 정립
+- LLM 백엔드를 MLX로 시작하되 OpenAI / Ollama 전환이 설정 변경만으로 가능해야 함
+- MySQL DB 연동이 예정되어 있어 DB 서비스 레이어 선제 설계 필요
+
+---
+
+## 2. Scope
+
+### 2.1 In Scope
+- [ ] IoC Container 설정 (`app/container.py`)
+- [ ] Strategy 패턴으로 LLM 백엔드 추상화 (`AbstractModelService` + `MlxModelService`)
+- [ ] `ChatService` — 대화 오케스트레이션
+- [ ] `HistoryService` — 히스토리 관리 (trim 포함)
+- [ ] `DatabaseService` — MySQL 연결 (mysqlclient)
+- [ ] Observer/EventBus — 스트림 토큰 이벤트 발행
+- [ ] `CliUiService` — CLI 입출력
+- [ ] `main.py` — 컨테이너 부트스트랩 진입점
+- [ ] `config.py` — 환경변수 기반 설정
+
+### 2.2 Out of Scope
+- 웹 API (FastAPI 등) 레이어
+- 인증/세션 관리
+- 테스트 코드 (별도 Phase)
+
+---
+
+## 3. Requirements
+
+### 3.1 Functional Requirements
+
+| ID | Requirement | Priority |
+|----|-------------|----------|
+| FR-01 | IoC Container에서 모든 서비스를 Singleton/Factory로 등록·해석 | High |
+| FR-02 | LLM 백엔드를 Strategy 인터페이스로 추상화, 구현체 교체 가능 | High |
+| FR-03 | ChatService가 HistoryService·ModelService·EventBus를 주입받아 동작 | High |
+| FR-04 | 스트림 토큰 출력을 Observer 패턴으로 처리 (핸들러 교체 가능) | Medium |
+| FR-05 | DatabaseService가 MySQL 연결을 캡슐화, 다른 서비스에서 주입받아 사용 | Medium |
+| FR-06 | 설정은 환경변수 + config.py에서 중앙 관리 | High |
+
+### 3.2 Non-Functional Requirements
+
+| Category | Criteria |
+|----------|----------|
+| 확장성 | 새 LLM 백엔드 추가 시 AbstractModelService 구현만으로 완성 |
+| 결합도 | 서비스 간 직접 import 없이 Container 통해서만 의존 |
+| 메모리 | MLX 모델은 Singleton으로 1회만 로딩 |
+
+---
+
+## 4. Success Criteria
+
+- [ ] `python -m app.main` 실행 시 기존 chat.py와 동일하게 동작
+- [ ] `MlxModelService`를 `MockModelService`로 교체해도 나머지 코드 변경 없음
+- [ ] `Container` 설정만 바꿔서 CLI → 다른 UI로 전환 가능한 구조
+
+---
+
+## 5. Risks
+
+| Risk | Impact | Likelihood | Mitigation |
+|------|--------|------------|------------|
+| MLX Singleton 초기화 시간 (~수초) | Medium | High | 로딩 중 진행 표시 유지 |
+| dependency-injector 버전 충돌 | Low | Low | requirements.txt 버전 고정 |
+| MySQL 연결 설정 누락 | Medium | Medium | 미설정 시 graceful skip |
+
+---
+
+## 6. Architecture Decisions
+
+| Decision | Selected | Rationale |
+|----------|----------|-----------|
+| IoC 라이브러리 | dependency-injector | DeclarativeContainer, Singleton/Factory 지원 |
+| LLM 추상화 | Strategy 패턴 | 백엔드 교체 시 코드 변경 최소화 |
+| 스트림 처리 | Observer/EventBus | 출력 핸들러(CLI, 파일, WebSocket)를 분리 |
+| DB | MySQL (mysqlclient) | 단순 Service로 캡슐화, 추후 ORM 추가 가능 |
+
+---
+
+## 7. Directory Structure
+
+```
+youlbot/
+├── app/
+│   ├── __init__.py
+│   ├── config.py              # 환경변수 기반 설정
+│   ├── container.py           # IoC DeclarativeContainer
+│   ├── main.py                # 부트스트랩 진입점
+│   └── services/
+│       ├── model/
+│       │   ├── base.py        # AbstractModelService (Strategy 인터페이스)
+│       │   └── mlx_model.py   # MlxModelService (Strategy 구현체)
+│       ├── chat/
+│       │   ├── chat_service.py
+│       │   └── history_service.py
+│       ├── db/
+│       │   └── mysql_service.py
+│       ├── ui/
+│       │   └── cli_service.py
+│       └── events/
+│           ├── event_bus.py   # Observer EventBus
+│           └── handlers.py    # StreamTokenHandler 등
+├── chat.py                    # 기존 파일 (보존)
+└── requirements.txt
+```
@@ -0,0 +1,51 @@
+from container import Container
+from services.chat.chat_service import ChatService
+
+
+def main() -> None:
+    container = Container()
+
+    ui = container.ui_service()
+    model = container.model_service()
+    bus = container.event_bus()
+    db = container.db_service()
+    repo = container.conversation_repository()
+
+    bus.subscribe(ChatService.EVENT_TOKEN, container.stream_token_handler())
+    bus.subscribe(ChatService.EVENT_END, container.stream_end_handler())
+
+    ui.show_banner(container.config().model_id)
+    model.load()
+    db.connect()
+    db.init_schema()
+
+    chat = container.chat_service()
+
+    while True:
+        try:
+            user_input = ui.prompt_user()
+        except (EOFError, KeyboardInterrupt):
+            print("\n대화를 종료합니다.")
+            break
+
+        if not user_input:
+            continue
+
+        if ui.is_exit_command(user_input):
+            print("대화를 종료합니다.")
+            break
+
+        if ui.is_reset_command(user_input):
+            repo.create_conversation()
+            chat = container.chat_service()
+            print("\n[대화가 초기화되었습니다.]\n")
+            continue
+
+        ui.show_assistant_prefix()
+        chat.respond(user_input)
+
+    db.close()
+
+
+if __name__ == "__main__":
+    main()
@@ -1 +1,4 @@
 mlx-lm>=0.19.0
+dependency-injector>=4.41.0
+PyMySQL>=1.1.0
+pydantic-settings>=2.0.0
@@ -0,0 +1,35 @@
+from services.model.base import AbstractModelService
+from services.chat.history_service import HistoryService
+from services.events.event_bus import EventBus
+
+
+class ChatService:
+    """대화 오케스트레이션 서비스."""
+
+    EVENT_TOKEN = "stream.token"
+    EVENT_END = "stream.end"
+
+    def __init__(
+        self,
+        model: AbstractModelService,
+        history: HistoryService,
+        event_bus: EventBus,
+        max_tokens: int,
+    ):
+        self._model = model
+        self._history = history
+        self._event_bus = event_bus
+        self._max_tokens = max_tokens
+
+    def respond(self, user_input: str) -> str:
+        self._history.add("user", user_input)
+        prompt = self._model.build_prompt(self._history.get())
+
+        response_text = ""
+        for token in self._model.stream(prompt, self._max_tokens):
+            self._event_bus.publish(self.EVENT_TOKEN, token)
+            response_text += token
+
+        self._event_bus.publish(self.EVENT_END)
+        self._history.add("assistant", response_text)
+        return response_text
@@ -0,0 +1,18 @@
+from services.model.base import AbstractModelService
+
+
+class CompactService:
+    """오래된 대화 턴을 LLM으로 요약하는 서비스."""
+
+    def __init__(self, model: AbstractModelService, max_tokens: int = 512):
+        self._model = model
+        self._max_tokens = max_tokens
+
+    def summarize(self, turns: list[dict]) -> str:
+        text = "\n".join(f"{t['role']}: {t['content']}" for t in turns)
+        prompt_history = [
+            {"role": "system", "content": "당신은 대화 요약 전문가입니다."},
+            {"role": "user", "content": f"다음 대화의 핵심 내용을 한국어로 간결하게 요약해주세요:\n\n{text}"},
+        ]
+        prompt = self._model.build_prompt(prompt_history)
+        return "".join(self._model.stream(prompt, self._max_tokens))
@@ -0,0 +1,81 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from services.db.conversation_repository import ConversationRepository
+    from services.chat.compact_service import CompactService
+
+
+class HistoryService:
+    """대화 히스토리를 관리하는 서비스."""
+
+    def __init__(
+        self,
+        system_prompt: str,
+        max_turns: int,
+        compact_threshold: int,
+        repository: ConversationRepository | None = None,
+        compact_service: CompactService | None = None,
+    ):
+        self._system_prompt = system_prompt
+        self._max_turns = max_turns
+        self._compact_threshold = compact_threshold
+        self._repository = repository
+        self._compact_service = compact_service
+        self._summary: str | None = None
+        self._turns: list[dict] = []
+        self._conversation_id: int | None = None
+
+        if repository:
+            self._load_or_create()
+
+    # ── DB 초기화 ────────────────────────────────────────────────
+
+    def _load_or_create(self) -> None:
+        conv_id = self._repository.get_latest_conversation_id()
+        if conv_id:
+            summary_id, summary = self._repository.get_latest_summary(conv_id)
+            turns = self._repository.load_turns_after(
+                conv_id, summary_id, self._compact_threshold * 2
+            )
+            self._summary = summary
+            self._turns = turns
+            self._conversation_id = conv_id
+        else:
+            self._conversation_id = self._repository.create_conversation()
+
+    # ── 공개 인터페이스 ───────────────────────────────────────────
+
+    def add(self, role: str, content: str) -> None:
+        self._turns.append({"role": role, "content": content})
+        if self._repository and self._conversation_id:
+            self._repository.save_message(self._conversation_id, role, content)
+        if role == "assistant":
+            self._maybe_compact()
+
+    def get(self) -> list[dict]:
+        msgs = [{"role": "system", "content": self._system_prompt}]
+        if self._summary:
+            msgs.append({"role": "system", "content": f"[이전 대화 요약]\n{self._summary}"})
+        msgs.extend(self._turns)
+        return msgs
+
+    def reset(self, new_conversation_id: int) -> None:
+        self._summary = None
+        self._turns = []
+        self._conversation_id = new_conversation_id
+
+    # ── 내부 ─────────────────────────────────────────────────────
+
+    def _maybe_compact(self) -> None:
+        if not self._compact_service or len(self._turns) <= self._compact_threshold:
+            return
+
+        mid = len(self._turns) // 2
+        old_turns, self._turns = self._turns[:mid], self._turns[mid:]
+
+        print("\n[대화 내용을 압축하는 중...]\n", flush=True)
+        self._summary = self._compact_service.summarize(old_turns)
+
+        if self._repository and self._conversation_id:
+            self._repository.save_summary(self._conversation_id, self._summary)
@@ -0,0 +1,64 @@
+from __future__ import annotations
+from services.db.mysql_service import DatabaseService
+
+
+class ConversationRepository:
+    """td_conversations / td_messages 테이블 접근을 담당하는 Repository."""
+
+    def __init__(self, db: DatabaseService):
+        self._db = db
+
+    def create_conversation(self) -> int:
+        return self._db.execute_write(
+            "INSERT INTO td_conversations () VALUES ()"
+        )
+
+    def get_latest_conversation_id(self) -> int | None:
+        rows = self._db.execute(
+            "SELECT id FROM td_conversations ORDER BY created_at DESC LIMIT 1"
+        )
+        return rows[0]["id"] if rows else None
+
+    def save_message(self, conversation_id: int, role: str, content: str) -> None:
+        self._db.execute_write(
+            "INSERT INTO td_messages (conversation_id, role, content) VALUES (%s, %s, %s)",
+            (conversation_id, role, content),
+        )
+
+    def save_summary(self, conversation_id: int, summary: str) -> None:
+        self._db.execute_write(
+            "INSERT INTO td_messages (conversation_id, role, content) VALUES (%s, %s, %s)",
+            (conversation_id, "summary", summary),
+        )
+
+    def get_latest_summary(self, conversation_id: int) -> tuple[int | None, str | None]:
+        """가장 최근 요약 메시지의 (id, content)를 반환. 없으면 (None, None)."""
+        rows = self._db.execute(
+            """SELECT id, content FROM td_messages
+               WHERE conversation_id = %s AND role = 'summary'
+               ORDER BY created_at DESC LIMIT 1""",
+            (conversation_id,),
+        )
+        if rows:
+            return rows[0]["id"], rows[0]["content"]
+        return None, None
+
+    def load_turns_after(
+        self, conversation_id: int, after_id: int | None, limit: int
+    ) -> list[dict]:
+        """요약 이후의 user/assistant 턴을 최근 limit개 반환."""
+        if after_id is not None:
+            rows = self._db.execute(
+                """SELECT role, content FROM td_messages
+                   WHERE conversation_id = %s AND id > %s AND role IN ('user', 'assistant')
+                   ORDER BY created_at DESC LIMIT %s""",
+                (conversation_id, after_id, limit),
+            )
+        else:
+            rows = self._db.execute(
+                """SELECT role, content FROM td_messages
+                   WHERE conversation_id = %s AND role IN ('user', 'assistant')
+                   ORDER BY created_at DESC LIMIT %s""",
+                (conversation_id, limit),
+            )
+        return list(reversed(rows))
@@ -0,0 +1,63 @@
+from __future__ import annotations
+from typing import Any
+
+
+class DatabaseService:
+    """MySQL 연결을 캡슐화하는 서비스. 미설정 시 graceful skip."""
+
+    def __init__(self, host: str, port: int, db: str, user: str, password: str):
+        self._config = dict(host=host, port=port, db=db, user=user, passwd=password)
+        self._conn = None
+
+    def connect(self) -> None:
+        if not self._config["user"]:
+            return
+        try:
+            import pymysql
+            self._conn = pymysql.connect(**self._config)
+        except Exception as e:
+            print(f"[DB] 연결 실패 (선택적 기능): {e}")
+
+    def execute(self, sql: str, params: tuple = ()) -> list[dict[str, Any]]:
+        if self._conn is None:
+            return []
+        cursor = self._conn.cursor()
+        cursor.execute(sql, params)
+        columns = [d[0] for d in cursor.description or []]
+        return [dict(zip(columns, row)) for row in cursor.fetchall()]
+
+    def execute_write(self, sql: str, params: tuple = ()) -> int:
+        """INSERT/UPDATE/DELETE 실행 후 lastrowid 반환."""
+        if self._conn is None:
+            return 0
+        cursor = self._conn.cursor()
+        cursor.execute(sql, params)
+        self._conn.commit()
+        return cursor.lastrowid
+
+    def init_schema(self) -> None:
+        if self._conn is None:
+            return
+        cursor = self._conn.cursor()
+        cursor.execute("""
+            CREATE TABLE IF NOT EXISTS td_conversations (
+                id INT AUTO_INCREMENT PRIMARY KEY,
+                created_at DATETIME DEFAULT CURRENT_TIMESTAMP
+            )
+        """)
+        cursor.execute("""
+            CREATE TABLE IF NOT EXISTS td_messages (
+                id INT AUTO_INCREMENT PRIMARY KEY,
+                conversation_id INT NOT NULL,
+                role VARCHAR(20) NOT NULL,
+                content TEXT NOT NULL,
+                created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+                FOREIGN KEY (conversation_id) REFERENCES td_conversations(id)
+            )
+        """)
+        self._conn.commit()
+
+    def close(self) -> None:
+        if self._conn:
+            self._conn.close()
+            self._conn = None
@@ -0,0 +1,19 @@
+from collections import defaultdict
+from typing import Callable
+
+
+class EventBus:
+    """Observer 패턴 기반 이벤트 버스."""
+
+    def __init__(self):
+        self._handlers: dict[str, list[Callable]] = defaultdict(list)
+
+    def subscribe(self, event: str, handler: Callable) -> None:
+        self._handlers[event].append(handler)
+
+    def unsubscribe(self, event: str, handler: Callable) -> None:
+        self._handlers[event].remove(handler)
+
+    def publish(self, event: str, *args, **kwargs) -> None:
+        for handler in self._handlers[event]:
+            handler(*args, **kwargs)
@@ -0,0 +1,15 @@
+import sys
+
+
+class StreamTokenHandler:
+    """스트림 토큰을 stdout에 실시간 출력하는 핸들러."""
+
+    def __call__(self, token: str) -> None:
+        print(token, end="", flush=True)
+
+
+class StreamEndHandler:
+    """스트림 종료 시 개행을 출력하는 핸들러."""
+
+    def __call__(self) -> None:
+        print("\n")
@@ -0,0 +1,18 @@
+from abc import ABC, abstractmethod
+from typing import Iterator
+
+
+class AbstractModelService(ABC):
+    """LLM 백엔드 Strategy 인터페이스."""
+
+    @abstractmethod
+    def load(self) -> None:
+        """모델을 메모리에 로드한다."""
+
+    @abstractmethod
+    def stream(self, prompt: str, max_tokens: int) -> Iterator[str]:
+        """프롬프트를 받아 토큰을 스트리밍한다."""
+
+    @abstractmethod
+    def build_prompt(self, history: list[dict]) -> str:
+        """대화 히스토리를 모델 입력 형식으로 변환한다."""
@@ -0,0 +1,29 @@
+from typing import Iterator
+
+from services.model.base import AbstractModelService
+
+
+class MlxModelService(AbstractModelService):
+    """MLX 기반 로컬 LLM Strategy 구현체."""
+
+    def __init__(self, model_id: str):
+        self._model_id = model_id
+        self._model = None
+        self._tokenizer = None
+
+    def load(self) -> None:
+        from mlx_lm import load
+        print(f"모델 로딩 중: {self._model_id}")
+        self._model, self._tokenizer = load(self._model_id)
+
+    def build_prompt(self, history: list[dict]) -> str:
+        return self._tokenizer.apply_chat_template(
+            history,
+            tokenize=False,
+            add_generation_prompt=True,
+        )
+
+    def stream(self, prompt: str, max_tokens: int) -> Iterator[str]:
+        from mlx_lm import stream_generate
+        for chunk in stream_generate(self._model, self._tokenizer, prompt=prompt, max_tokens=max_tokens):
+            yield chunk.text
@@ -0,0 +1,23 @@
+class CliUiService:
+    """CLI 입출력 서비스."""
+
+    def show_banner(self, model_id: str) -> None:
+        print(f"모델 로딩 중: {model_id}")
+        print("(첫 실행 시 HuggingFace에서 자동 다운로드됩니다. 약 4.5GB)\n")
+        print("=" * 50)
+        print("육아 & 금융 상담 챗봇 시작!")
+        print("종료: '종료' / 'quit' / 'exit' 입력")
+        print("초기화: 'reset' 또는 'clear' 입력")
+        print("=" * 50 + "\n")
+
+    def prompt_user(self) -> str:
+        return input("나: ").strip()
+
+    def show_assistant_prefix(self) -> None:
+        print("\n도우미: ", end="", flush=True)
+
+    def is_exit_command(self, text: str) -> bool:
+        return text.lower() in ("종료", "quit", "exit")
+
+    def is_reset_command(self, text: str) -> bool:
+        return text.lower() in ("reset", "clear", "초기화")