diff --git a/.env.example b/.env.example index 71ab8d9..52a7ef6 100644 --- a/.env.example +++ b/.env.example @@ -57,6 +57,10 @@ ADMIN_API_KEY= # Anthropic API Key (AI Coach 프록시, 미설정 시 AI Coach 비활성화) ANTHROPIC_API_KEY= +# Ollama 서버 (Windows AI PC의 Ollama 엔드포인트) — 뉴스 요약용 +OLLAMA_URL=http://192.168.45.59:11435 +OLLAMA_MODEL=qwen3:14b + # [BLOG LAB] # Naver Search API (https://developers.naver.com 에서 발급) NAVER_CLIENT_ID= diff --git a/agent-office/app/agents/stock.py b/agent-office/app/agents/stock.py index eef5caa..d6de585 100644 --- a/agent-office/app/agents/stock.py +++ b/agent-office/app/agents/stock.py @@ -14,22 +14,35 @@ class StockAgent(BaseAgent): return task_id = create_task(self.agent_id, "news_summary", {"limit": 15}) - await self.transition("working", "뉴스 수집 중...", task_id) + await self.transition("working", "AI 뉴스 요약 생성 중...", task_id) try: - news = await service_proxy.fetch_stock_news(limit=15) - indices = await service_proxy.fetch_stock_indices() - - summary = self._format_news_summary(news, indices) + # AI 요약 호출 (뉴스 수집 + LLM 처리는 stock-lab이 담당) + result = await service_proxy.summarize_stock_news(limit=15) await self.transition("reporting", "뉴스 요약 전송 중...") - from ..telegram_bot import send_stock_summary - tg_result = await send_stock_summary(summary) + # 새 통합 텔레그램 API 사용 + from ..telegram import send_agent_message + tg_result = await send_agent_message( + agent_id=self.agent_id, + kind="report", + title="아침 시장 브리핑", + body=result["summary"], + task_id=task_id, + metadata={ + "tokens": result["tokens"]["total"], + "duration_ms": result["duration_ms"], + "model": result["model"], + }, + ) update_task_status(task_id, "succeeded", { - "summary": summary, - "news_count": len(news) if isinstance(news, list) else 0, + "summary": result["summary"], + "article_count": result.get("article_count", 0), + "tokens": result["tokens"], + "model": result["model"], + "duration_ms": result["duration_ms"], "telegram_sent": tg_result.get("ok", False), "telegram_message_id": tg_result.get("message_id"), }) @@ -50,8 +63,13 @@ class StockAgent(BaseAgent): async def on_command(self, command: str, params: dict) -> dict: if command == "test_telegram": - from ..telegram_bot import send_message - result = await send_message("🔔 [주식 에이전트] 텔레그램 테스트 메시지입니다.") + from ..telegram import send_agent_message + result = await send_agent_message( + agent_id=self.agent_id, + kind="info", + title="연결 테스트", + body="텔레그램 연동이 정상적으로 동작합니다.", + ) return { "ok": result.get("ok", False), "message": "텔레그램 전송 성공" if result.get("ok") else "텔레그램 전송 실패", @@ -88,30 +106,3 @@ class StockAgent(BaseAgent): async def on_approval(self, task_id: str, approved: bool, feedback: str = "") -> None: pass - - def _format_news_summary(self, news, indices) -> str: - lines = ["📈 [주식 에이전트] 아침 뉴스 요약", "━" * 20] - - if isinstance(news, list): - for item in news[:10]: - title = item.get("title", "") - if title: - lines.append(f"• {title}") - elif isinstance(news, dict) and "articles" in news: - for item in news["articles"][:10]: - title = item.get("title", "") - if title: - lines.append(f"• {title}") - - if indices: - lines.append("") - lines.append("📊 주요 지수") - if isinstance(indices, dict): - for key, val in indices.items(): - if isinstance(val, dict): - name = val.get("name", key) - price = val.get("price", "") - change = val.get("change", "") - lines.append(f"{name}: {price} ({change})") - - return "\n".join(lines) diff --git a/agent-office/app/db.py b/agent-office/app/db.py index 850a83f..01b9aab 100644 --- a/agent-office/app/db.py +++ b/agent-office/app/db.py @@ -261,6 +261,59 @@ def mark_telegram_responded(callback_id: str, action: str) -> None: ) +def get_token_usage_stats(agent_id: str, days: int = 1) -> dict: + """지정 에이전트의 최근 N일 토큰 사용량 집계. + + agent_tasks 테이블의 result_data JSON에서 tokens.total을 합산. + 반환: {"total_tokens": int, "task_count": int, "by_day": [{"date": "YYYY-MM-DD", "tokens": int}]} + """ + with _conn() as conn: + rows = conn.execute( + """ + SELECT completed_at, result_data + FROM agent_tasks + WHERE agent_id = ? + AND status = 'succeeded' + AND completed_at IS NOT NULL + AND completed_at >= strftime('%Y-%m-%dT%H:%M:%fZ','now', ?) + """, + (agent_id, f"-{int(days)} days"), + ).fetchall() + + total_tokens = 0 + task_count = 0 + by_day_map: Dict[str, int] = {} + for r in rows: + result_data = r["result_data"] + if not result_data: + continue + try: + parsed = json.loads(result_data) + except Exception: + continue + tokens = parsed.get("tokens") if isinstance(parsed, dict) else None + total = 0 + if isinstance(tokens, dict): + total = int(tokens.get("total", 0) or 0) + if total <= 0: + continue + total_tokens += total + task_count += 1 + completed_at = r["completed_at"] or "" + day = completed_at[:10] if completed_at else "unknown" + by_day_map[day] = by_day_map.get(day, 0) + total + + by_day = [ + {"date": d, "tokens": t} + for d, t in sorted(by_day_map.items()) + ] + return { + "total_tokens": total_tokens, + "task_count": task_count, + "by_day": by_day, + } + + def get_activity_feed(limit: int = 50, offset: int = 0) -> dict: with _conn() as conn: total_row = conn.execute(""" diff --git a/agent-office/app/main.py b/agent-office/app/main.py index c404828..875357b 100644 --- a/agent-office/app/main.py +++ b/agent-office/app/main.py @@ -138,10 +138,26 @@ async def approve(body: ApprovalRequest): # --- Telegram Webhook --- +async def _agent_dispatcher(agent_id: str, command: str, params: dict) -> dict: + """텔레그램 라우터가 호출하는 에이전트 디스패처.""" + # 전역 상태 조회 + if agent_id == "__global__" and command == "status": + result = {} + for aid, agent in AGENT_REGISTRY.items(): + result[aid] = {"state": agent.state, "detail": agent.state_detail} + return result + + agent = AGENT_REGISTRY.get(agent_id) + if agent is None: + return {"ok": False, "message": f"Unknown agent: {agent_id}"} + return await agent.on_command(command, params or {}) + + @app.post("/api/agent-office/telegram/webhook") async def telegram_webhook(data: dict): - result = await telegram_bot.handle_webhook(data) - if result: + result = await telegram_bot.handle_webhook(data, agent_dispatcher=_agent_dispatcher) + # callback_query (승인/거절) → 기존 승인 흐름 + if result and "approved" in result: agent = get_agent(result["agent_id"]) if agent: await agent.on_approval(result["task_id"], result["approved"]) @@ -151,6 +167,11 @@ async def telegram_webhook(data: dict): def all_states(): return {"agents": get_all_agent_states()} +@app.get("/api/agent-office/agents/{agent_id}/token-usage") +def agent_token_usage(agent_id: str, days: int = 1): + from .db import get_token_usage_stats + return get_token_usage_stats(agent_id, days) + @app.get("/api/agent-office/activity") def activity_feed(limit: int = 50, offset: int = 0): return get_activity_feed(limit, offset) diff --git a/agent-office/app/service_proxy.py b/agent-office/app/service_proxy.py index ae6d3b6..c55b2d2 100644 --- a/agent-office/app/service_proxy.py +++ b/agent-office/app/service_proxy.py @@ -18,6 +18,18 @@ async def fetch_stock_indices() -> Dict[str, Any]: resp.raise_for_status() return resp.json() +async def summarize_stock_news(limit: int = 15) -> Dict[str, Any]: + """stock-lab의 AI 요약 엔드포인트 호출. + 반환: {"summary": str, "tokens": {...}, "model": str, "duration_ms": int, "article_count": int} + """ + async with httpx.AsyncClient(timeout=90.0) as client: + resp = await client.post( + f"{STOCK_LAB_URL}/api/stock/news/summarize", + json={"limit": limit}, + ) + resp.raise_for_status() + return resp.json() + async def generate_music(payload: dict) -> Dict[str, Any]: resp = await _client.post(f"{MUSIC_LAB_URL}/api/music/generate", json=payload) resp.raise_for_status() diff --git a/agent-office/app/telegram/__init__.py b/agent-office/app/telegram/__init__.py new file mode 100644 index 0000000..7bb1ce7 --- /dev/null +++ b/agent-office/app/telegram/__init__.py @@ -0,0 +1,19 @@ +"""Telegram 통합 메시지 패키지.""" +from .agent_registry import AGENT_META, get_agent_meta, register_agent +from .messaging import send_agent_message, send_approval_request, send_raw +from .router import parse_command, resolve_agent_command, HELP_TEXT +from .webhook import handle_webhook, setup_webhook + +__all__ = [ + "send_agent_message", + "send_approval_request", + "send_raw", + "handle_webhook", + "setup_webhook", + "get_agent_meta", + "register_agent", + "AGENT_META", + "parse_command", + "resolve_agent_command", + "HELP_TEXT", +] diff --git a/agent-office/app/telegram/agent_registry.py b/agent-office/app/telegram/agent_registry.py new file mode 100644 index 0000000..3ac9478 --- /dev/null +++ b/agent-office/app/telegram/agent_registry.py @@ -0,0 +1,30 @@ +"""에이전트 메타 등록소.""" + +AGENT_META = { + "stock": { + "display_name": "주식 트레이더", + "emoji": "📈", + "color": "#4488cc", + }, + "music": { + "display_name": "음악 프로듀서", + "emoji": "🎵", + "color": "#44aa88", + }, +} + + +def get_agent_meta(agent_id: str) -> dict: + return AGENT_META.get( + agent_id, + {"display_name": agent_id, "emoji": "🤖", "color": "#888"}, + ) + + +def register_agent(agent_id: str, display_name: str, emoji: str, color: str = "#888"): + """향후 에이전트 동적 등록용""" + AGENT_META[agent_id] = { + "display_name": display_name, + "emoji": emoji, + "color": color, + } diff --git a/agent-office/app/telegram/client.py b/agent-office/app/telegram/client.py new file mode 100644 index 0000000..6ebf7a9 --- /dev/null +++ b/agent-office/app/telegram/client.py @@ -0,0 +1,18 @@ +"""Telegram Bot API 저수준 래퍼.""" +import httpx + +from ..config import TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, TELEGRAM_WEBHOOK_URL + +_BASE = "https://api.telegram.org/bot" + + +def _enabled() -> bool: + return bool(TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID) + + +async def api_call(method: str, payload: dict) -> dict: + if not _enabled(): + return {"ok": False, "description": "Telegram not configured"} + async with httpx.AsyncClient(timeout=10.0) as client: + resp = await client.post(f"{_BASE}{TELEGRAM_BOT_TOKEN}/{method}", json=payload) + return resp.json() diff --git a/agent-office/app/telegram/formatter.py b/agent-office/app/telegram/formatter.py new file mode 100644 index 0000000..0fc9b4e --- /dev/null +++ b/agent-office/app/telegram/formatter.py @@ -0,0 +1,43 @@ +"""에이전트 메시지 포맷팅.""" +from typing import Literal, Optional + +from .agent_registry import get_agent_meta + +MessageKind = Literal["report", "alert", "approval", "error", "info"] + +KIND_ICONS = { + "report": "📊", + "alert": "🔔", + "approval": "✋", + "error": "⚠️", + "info": "ℹ️", +} + + +def format_agent_message( + agent_id: str, + kind: MessageKind, + title: str, + body: str, + metadata: Optional[dict] = None, +) -> str: + meta = get_agent_meta(agent_id) + icon = KIND_ICONS.get(kind, "") + header = f"{icon} [{meta['emoji']} {meta['display_name']}] {title}" + + lines = [header, "━" * 20, body] + + if metadata: + footer_parts = [] + if "tokens" in metadata: + footer_parts.append(f"🧮 {metadata['tokens']:,} tokens") + if "duration_ms" in metadata: + seconds = metadata["duration_ms"] / 1000 + footer_parts.append(f"⏱ {seconds:.1f}s") + if "model" in metadata: + footer_parts.append(f"🤖 {metadata['model']}") + if footer_parts: + lines.append("") + lines.append(f"{' · '.join(footer_parts)}") + + return "\n".join(lines) diff --git a/agent-office/app/telegram/messaging.py b/agent-office/app/telegram/messaging.py new file mode 100644 index 0000000..7d13636 --- /dev/null +++ b/agent-office/app/telegram/messaging.py @@ -0,0 +1,68 @@ +"""고수준 메시지 전송 API.""" +import uuid +from typing import Optional + +from ..config import TELEGRAM_CHAT_ID +from ..db import save_telegram_callback +from .client import _enabled, api_call +from .formatter import MessageKind, format_agent_message + + +async def send_raw(text: str, reply_markup: Optional[dict] = None) -> dict: + """가장 저수준. 원문 텍스트 그대로 전송.""" + if not _enabled(): + return {"ok": False, "message_id": None} + payload = { + "chat_id": TELEGRAM_CHAT_ID, + "text": text, + "parse_mode": "HTML", + } + if reply_markup: + payload["reply_markup"] = reply_markup + result = await api_call("sendMessage", payload) + return { + "ok": result.get("ok", False), + "message_id": result.get("result", {}).get("message_id") if result.get("ok") else None, + } + + +async def send_agent_message( + agent_id: str, + kind: MessageKind, + title: str, + body: str, + task_id: Optional[str] = None, + actions: Optional[list] = None, + metadata: Optional[dict] = None, +) -> dict: + """통합 에이전트 메시지 API. 모든 에이전트가 이걸 씀.""" + text = format_agent_message(agent_id, kind, title, body, metadata) + reply_markup = None + if actions: + buttons = [] + for action in actions: + cb_id = f"{action['action']}_{uuid.uuid4().hex[:8]}" + save_telegram_callback(cb_id, task_id or "", agent_id) + buttons.append({"text": action["label"], "callback_data": cb_id}) + reply_markup = {"inline_keyboard": [buttons]} + return await send_raw(text, reply_markup) + + +async def send_approval_request( + agent_id: str, + task_id: str, + title: str, + detail: str, +) -> dict: + """승인/거절 단축 헬퍼.""" + return await send_agent_message( + agent_id=agent_id, + kind="approval", + title=title, + body=detail, + task_id=task_id, + actions=[ + {"label": "✅ 승인", "action": "approve"}, + {"label": "❌ 거절", "action": "reject"}, + ], + ) diff --git a/agent-office/app/telegram/router.py b/agent-office/app/telegram/router.py new file mode 100644 index 0000000..4c200c8 --- /dev/null +++ b/agent-office/app/telegram/router.py @@ -0,0 +1,87 @@ +"""텔레그램 메시지 명령 → 에이전트 라우팅. +새 명령을 추가하려면 AGENT_COMMAND_MAP에 등록만 하면 됨.""" +from typing import Optional + + +def parse_command(text: str) -> Optional[tuple]: + """슬래시 명령 파싱. + + 반환: (agent_id_or_None, command, args_list) 또는 None + + 예시: + /stock news -> ("stock", "news", []) + /status -> (None, "status", []) + /music compose 잔잔한 피아노 -> ("music", "compose", ["잔잔한 피아노"]) + """ + if not text: + return None + text = text.strip() + if not text.startswith("/"): + return None + parts = text[1:].split(maxsplit=2) + if not parts: + return None + + first = parts[0].lower() + + # 전역 명령 + if first in ("status", "agents", "help"): + return (None, first, parts[1:] if len(parts) > 1 else []) + + # 에이전트 명령: / [args...] + if len(parts) < 2: + return None + + agent_id = first + command = parts[1].lower() + args = [parts[2]] if len(parts) > 2 else [] + return (agent_id, command, args) + + +# 에이전트별 텔레그램 → 내부 command 매핑 +# 텔레그램에서 친숙한 이름 -> (실제 on_command의 command, 기본 params) +AGENT_COMMAND_MAP = { + "stock": { + "news": ("fetch_news", {}), + "alerts": ("list_alerts", {}), + "test": ("test_telegram", {}), + }, + "music": { + "credits": ("credits", {}), + # compose는 인자 필요 — 아래 특수 케이스에서 처리 + }, +} + + +def resolve_agent_command(agent_id: str, command: str, args: list) -> Optional[tuple]: + """(internal_command, params) 반환. 매핑 없으면 None.""" + mapping = AGENT_COMMAND_MAP.get(agent_id, {}).get(command) + if mapping is None: + # 특수 케이스: music compose + if agent_id == "music" and command == "compose" and args: + return ("compose", {"prompt": " ".join(args)}) + return None + internal_cmd, base_params = mapping + params = dict(base_params) + if args: + # args가 있으면 첫 번째(합쳐진 나머지)를 message로 자동 주입 + params["message"] = " ".join(args) + return (internal_cmd, params) + + +HELP_TEXT = """🤖 Agent Office 텔레그램 명령 + +전역 +/status — 모든 에이전트 상태 +/agents — 에이전트 목록 +/help — 이 도움말 + +📈 주식 트레이더 +/stock news — 뉴스 AI 요약 실행 +/stock alerts — 알람 목록 +/stock test — 텔레그램 테스트 + +🎵 음악 프로듀서 +/music credits — Suno 크레딧 조회 +/music compose <프롬프트> — 작곡 시작 +""" diff --git a/agent-office/app/telegram/webhook.py b/agent-office/app/telegram/webhook.py new file mode 100644 index 0000000..a0ae73c --- /dev/null +++ b/agent-office/app/telegram/webhook.py @@ -0,0 +1,149 @@ +"""텔레그램 Webhook 이벤트 처리.""" +from typing import Optional + +from ..db import get_telegram_callback, mark_telegram_responded +from .client import _enabled, api_call + + +async def handle_webhook(data: dict, agent_dispatcher=None) -> Optional[dict]: + """텔레그램에서 들어오는 이벤트 처리. + + - callback_query(인라인 버튼)는 항상 처리 → 승인/거절 dict 반환 + - message(텍스트 슬래시 명령)는 `agent_dispatcher`가 주입된 경우에만 처리 + + agent_dispatcher: async (agent_id, command, params) -> dict + - agent_id == "__global__", command == "status" 특수 케이스는 + {agent_id: {state, detail}} dict를 반환해야 함. + """ + callback_query = data.get("callback_query") + if callback_query: + return await _handle_callback(callback_query) + + message = data.get("message") + if message and message.get("text") and agent_dispatcher is not None: + return await _handle_message(message, agent_dispatcher) + + return None + + +async def _handle_callback(callback_query: dict) -> Optional[dict]: + """기존 승인/거절 콜백 처리 로직.""" + callback_id = callback_query.get("data", "") + cb = get_telegram_callback(callback_id) + if not cb: + return None + + action = callback_id.split("_")[0] + mark_telegram_responded(callback_id, action) + + feedback_text = { + "approve": "승인됨 ✅", + "reject": "거절됨 ❌", + }.get(action, f"처리됨: {action}") + + await api_call( + "answerCallbackQuery", + { + "callback_query_id": callback_query["id"], + "text": feedback_text, + }, + ) + + return { + "task_id": cb["task_id"], + "agent_id": cb["agent_id"], + "action": action, + "approved": action == "approve", + } + + +async def _handle_message(message: dict, agent_dispatcher) -> Optional[dict]: + """슬래시 명령 메시지 처리.""" + from .router import parse_command, resolve_agent_command, HELP_TEXT + from .messaging import send_raw, send_agent_message + from .agent_registry import AGENT_META + + text = message.get("text", "") + parsed = parse_command(text) + if not parsed: + return None + + agent_id, command, args = parsed + + # 전역 명령 + if agent_id is None: + if command == "help": + await send_raw(HELP_TEXT) + return {"handled": "help"} + + if command == "agents": + lines = ["📋 등록된 에이전트", ""] + for aid, meta in AGENT_META.items(): + lines.append( + f"{meta['emoji']} {meta['display_name']} /{aid}" + ) + await send_raw("\n".join(lines)) + return {"handled": "agents"} + + if command == "status": + try: + result = await agent_dispatcher("__global__", "status", {}) + body_lines = [] + if isinstance(result, dict): + for aid, info in result.items(): + meta = AGENT_META.get( + aid, {"emoji": "🤖", "display_name": aid} + ) + state = info.get("state", "unknown") if isinstance(info, dict) else "unknown" + body_lines.append( + f"{meta['emoji']} {meta['display_name']}: {state}" + ) + detail = info.get("detail") if isinstance(info, dict) else None + if detail: + body_lines.append(f" └ {detail}") + await send_raw("📊 전체 상태\n\n" + "\n".join(body_lines)) + except Exception as e: + await send_raw(f"⚠️ 상태 조회 실패: {e}") + return {"handled": "status"} + + return None + + # 에이전트 명령 + if agent_id not in AGENT_META: + await send_raw( + f"⚠️ 알 수 없는 에이전트: {agent_id}\n/help 로 사용 가능한 명령 확인" + ) + return {"handled": "unknown_agent"} + + resolved = resolve_agent_command(agent_id, command, args) + if resolved is None: + await send_raw( + f"⚠️ {agent_id}에서 {command} 명령은 지원하지 않습니다." + ) + return {"handled": "unknown_command"} + + internal_cmd, params = resolved + + try: + result = await agent_dispatcher(agent_id, internal_cmd, params) + ok = result.get("ok", False) if isinstance(result, dict) else False + msg = result.get("message", "") if isinstance(result, dict) else str(result) + + await send_agent_message( + agent_id=agent_id, + kind="info" if ok else "error", + title=f"{internal_cmd} 실행 결과", + body=msg or str(result), + ) + except Exception as e: + await send_raw(f"⚠️ 명령 실행 실패: {e}") + + return {"handled": "command", "agent_id": agent_id, "command": internal_cmd} + + +async def setup_webhook() -> dict: + from ..config import TELEGRAM_WEBHOOK_URL + + if not _enabled() or not TELEGRAM_WEBHOOK_URL: + return {"ok": False, "description": "Webhook URL not configured"} + return await api_call("setWebhook", {"url": TELEGRAM_WEBHOOK_URL}) diff --git a/agent-office/app/telegram_bot.py b/agent-office/app/telegram_bot.py index edacf9c..b165adc 100644 --- a/agent-office/app/telegram_bot.py +++ b/agent-office/app/telegram_bot.py @@ -1,86 +1,27 @@ -import json -import uuid -import httpx -from typing import Optional +"""Deprecated: app.telegram 패키지 사용 권장. 하위 호환용 re-export.""" +from .telegram import handle_webhook, send_approval_request, send_raw, setup_webhook +from .telegram.messaging import send_agent_message -from .config import TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, TELEGRAM_WEBHOOK_URL -from .db import save_telegram_callback, get_telegram_callback, mark_telegram_responded - -_BASE = "https://api.telegram.org/bot" - -def _enabled() -> bool: - return bool(TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID) - -async def _api(method: str, payload: dict) -> dict: - if not _enabled(): - return {"ok": False, "description": "Telegram not configured"} - async with httpx.AsyncClient(timeout=10.0) as client: - resp = await client.post(f"{_BASE}{TELEGRAM_BOT_TOKEN}/{method}", json=payload) - return resp.json() +# 기존 호출자가 쓰던 이름들 async def send_message(text: str, reply_markup: dict = None) -> dict: - payload = { - "chat_id": TELEGRAM_CHAT_ID, - "text": text, - "parse_mode": "HTML", - } - if reply_markup: - payload["reply_markup"] = reply_markup - result = await _api("sendMessage", payload) - return { - "ok": result.get("ok", False), - "message_id": result.get("result", {}).get("message_id") if result.get("ok") else None, - } + return await send_raw(text, reply_markup) + async def send_stock_summary(summary: str) -> dict: - return await send_message(summary) + return await send_raw(summary) -async def send_approval_request(agent_id: str, task_id: str, title: str, detail: str) -> dict: - approve_id = f"approve_{uuid.uuid4().hex[:8]}" - reject_id = f"reject_{uuid.uuid4().hex[:8]}" - - save_telegram_callback(approve_id, task_id, agent_id) - save_telegram_callback(reject_id, task_id, agent_id) - - text = f"{title}\n{'━' * 20}\n{detail}" - reply_markup = { - "inline_keyboard": [[ - {"text": "✅ 승인", "callback_data": approve_id}, - {"text": "❌ 거절", "callback_data": reject_id}, - ]] - } - return await send_message(text, reply_markup) async def send_task_result(agent_id: str, title: str, result: str) -> dict: - text = f"{title}\n{'━' * 20}\n{result}" - return await send_message(text) + return await send_agent_message(agent_id, "report", title, result) -async def handle_webhook(data: dict) -> Optional[dict]: - callback_query = data.get("callback_query") - if not callback_query: - return None - callback_id = callback_query.get("data", "") - cb = get_telegram_callback(callback_id) - if not cb: - return None - - action = "approve" if callback_id.startswith("approve_") else "reject" - mark_telegram_responded(callback_id, action) - - await _api("answerCallbackQuery", { - "callback_query_id": callback_query["id"], - "text": "승인됨 ✅" if action == "approve" else "거절됨 ❌", - }) - - return { - "task_id": cb["task_id"], - "agent_id": cb["agent_id"], - "action": action, - "approved": action == "approve", - } - -async def setup_webhook() -> dict: - if not _enabled() or not TELEGRAM_WEBHOOK_URL: - return {"ok": False, "description": "Webhook URL not configured"} - return await _api("setWebhook", {"url": TELEGRAM_WEBHOOK_URL}) +__all__ = [ + "send_message", + "send_stock_summary", + "send_task_result", + "send_approval_request", + "send_agent_message", + "handle_webhook", + "setup_webhook", +] diff --git a/docker-compose.yml b/docker-compose.yml index 69c821a..a1bbaba 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -38,6 +38,8 @@ services: - GEMINI_MODEL=${GEMINI_MODEL:-gemini-1.5-flash} - ADMIN_API_KEY=${ADMIN_API_KEY:-} - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY:-} + - OLLAMA_URL=${OLLAMA_URL:-http://192.168.45.59:11435} + - OLLAMA_MODEL=${OLLAMA_MODEL:-qwen3:14b} - CORS_ALLOW_ORIGINS=${CORS_ALLOW_ORIGINS:-http://localhost:3007,http://localhost:8080} volumes: - ${RUNTIME_PATH}/data/stock:/app/data diff --git a/stock-lab/app/ai_summarizer.py b/stock-lab/app/ai_summarizer.py new file mode 100644 index 0000000..bacadef --- /dev/null +++ b/stock-lab/app/ai_summarizer.py @@ -0,0 +1,113 @@ +"""Ollama 기반 뉴스 요약 모듈. + +Windows AI 서버(192.168.45.59:11435)의 Ollama에 연결하여 +한국어 시장 뉴스를 요약한다. 기존 WINDOWS_AI_SERVER_URL(KIS 래퍼)과는 +별개 경로이며, 본 모듈은 Ollama HTTP API(`/api/generate`)만 호출한다. +""" +import os +import logging +import time +from typing import List, Dict, Any + +import httpx + +logger = logging.getLogger("stock-lab.ai_summarizer") + +OLLAMA_URL = os.getenv("OLLAMA_URL", "http://192.168.45.59:11435") +OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "qwen3:14b") + +_PROMPT_TEMPLATE = """당신은 한국 주식 시장 애널리스트입니다. 아래 뉴스 목록을 읽고 투자자 관점에서 한국어로 간결하게 요약하세요. + +반드시 아래 형식을 그대로 지켜서 출력하세요. 다른 설명이나 서두, `` 같은 태그는 절대 출력하지 마세요. + +📌 시장 흐름 +(2줄 요약) + +🔥 주목 이슈 +• (이슈 1) +• (이슈 2) +• (이슈 3) + +💡 투자 관점 +(1줄 인사이트) + +=== 뉴스 목록 === +{news_block} +""" + + +class OllamaError(RuntimeError): + """Ollama 서버 호출 실패.""" + + +def _build_news_block(articles: List[Dict[str, Any]]) -> str: + lines = [] + for i, art in enumerate(articles, start=1): + title = (art.get("title") or "").strip() + content = (art.get("content") or art.get("summary") or "").strip() + if content: + lines.append(f"{i}. {title} — {content}") + else: + lines.append(f"{i}. {title}") + return "\n".join(lines) if lines else "(뉴스 없음)" + + +async def summarize_news(articles: List[Dict[str, Any]]) -> Dict[str, Any]: + """뉴스 리스트를 Ollama로 요약. + + Returns: + { + "summary": str, + "tokens": {"prompt": int, "completion": int, "total": int}, + "model": str, + "duration_ms": int, + } + Raises: + OllamaError: Ollama 호출 실패 시. + """ + prompt = _PROMPT_TEMPLATE.format(news_block=_build_news_block(articles)) + + url = f"{OLLAMA_URL.rstrip('/')}/api/generate" + payload = { + "model": OLLAMA_MODEL, + "prompt": prompt, + "stream": False, + } + + started = time.monotonic() + try: + async with httpx.AsyncClient(timeout=60.0) as client: + resp = await client.post(url, json=payload) + except httpx.HTTPError as e: + logger.error(f"Ollama 연결 실패 ({url}): {e}") + raise OllamaError(f"Ollama 연결 실패: {e}") from e + + if resp.status_code != 200: + logger.error(f"Ollama 응답 오류 {resp.status_code}: {resp.text[:200]}") + raise OllamaError(f"Ollama HTTP {resp.status_code}: {resp.text[:200]}") + + try: + data = resp.json() + except ValueError as e: + raise OllamaError(f"Ollama 응답 JSON 파싱 실패: {e}") from e + + summary = (data.get("response") or "").strip() + prompt_tokens = int(data.get("prompt_eval_count") or 0) + completion_tokens = int(data.get("eval_count") or 0) + # total_duration은 나노초 단위 + total_duration_ns = int(data.get("total_duration") or 0) + if total_duration_ns > 0: + duration_ms = total_duration_ns // 1_000_000 + else: + duration_ms = int((time.monotonic() - started) * 1000) + + return { + "summary": summary, + "tokens": { + "prompt": prompt_tokens, + "completion": completion_tokens, + "total": prompt_tokens + completion_tokens, + }, + "model": data.get("model") or OLLAMA_MODEL, + "duration_ms": duration_ms, + } diff --git a/stock-lab/app/main.py b/stock-lab/app/main.py index fea0eb0..69397b3 100644 --- a/stock-lab/app/main.py +++ b/stock-lab/app/main.py @@ -23,6 +23,7 @@ from .db import ( ) from .scraper import fetch_market_news, fetch_major_indices from .price_fetcher import get_current_prices +from .ai_summarizer import summarize_news, OllamaError app = FastAPI() @@ -144,6 +145,33 @@ def trigger_scrap(): run_scraping_job() return {"ok": True} + +class NewsSummarizeRequest(BaseModel): + limit: Optional[int] = 10 + + +@app.post("/api/stock/news/summarize") +async def summarize_latest_news(req: NewsSummarizeRequest = NewsSummarizeRequest()): + """최근 뉴스를 Ollama(qwen3:14b)로 요약""" + limit = req.limit if (req and req.limit) else 10 + articles = get_latest_articles(limit) + if not articles: + raise HTTPException(status_code=404, detail="요약할 뉴스가 없습니다.") + + try: + result = await summarize_news(articles) + except OllamaError as e: + logger.error(f"뉴스 요약 실패: {e}") + raise HTTPException(status_code=500, detail=f"Ollama 호출 실패: {e}") + except Exception as e: + logger.exception("뉴스 요약 중 예상치 못한 오류") + raise HTTPException(status_code=500, detail=f"뉴스 요약 실패: {e}") + + return { + **result, + "article_count": len(articles), + } + # --- Trading API (Windows Proxy, 인증 필요) --- @app.get("/api/trade/balance", dependencies=[Depends(verify_admin)]) diff --git a/stock-lab/requirements.txt b/stock-lab/requirements.txt index 048f02b..d6513f0 100644 --- a/stock-lab/requirements.txt +++ b/stock-lab/requirements.txt @@ -1,5 +1,6 @@ # 주식 서비스용 라이브러리 requests==2.32.3 +httpx==0.27.2 beautifulsoup4==4.12.3 fastapi==0.115.6 uvicorn[standard]==0.30.6