SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day49_AI Agent 구현 실습 2

Coding Agent

질문 - LLM - 답변이 아닌 생성 → 실행 → 검증 → 수정을 반복하기

MAX_RETRY = 5

for i in range(MAX_RETRY):

    code = llm.invoke(prompt)

    result = execute(code)

    if result.success:
        break

    prompt += f"""

에러

{result.stderr}

수정해라.
"""

 

 

 

재귀 Loop(Self-Correction)

에러가 발생하면 다시 LLM에게 준다.

while retry<5:

    code = llm.invoke(prompt)

    result = execute(code)

    if result.success:

        break

    prompt += result.stderr

 

 

 

 

Silent Failure

문자열비교등이 Silent Wrong Answer. 에러가 나지않는다고 정답이 아님으로 실행뿐만 아니라 결과도 검증해야한다.

expected = "30"

output = execute(code)

if output.stdout == expected:

    success = True

 

 

 

 

subprocess 격리 실행

LLM이 만든 코드는 exec() 실행하지않는다. 다른 코드들도 실행될 수 있기 때문이며 별도 프로세스에서 실행한다.

import subprocess

result = subprocess.run(

    ["python","solution.py"],

    capture_output=True,

    text=True

)

print(result.stdout)

 

 

 

 

 

확장 Loop

실제 Agent는 실행만 하지않고

질문 - 계획 - 코드작성 - 실행 - 테스트 - 실패 - 디버깅 - 다시작성 - git 저장

등..

 

 

 

전체예제

사용자 요청
      │
      ▼
  계획(Planner)
      │
      ▼
  코드 생성(LLM)
      │
      ▼
 subprocess 격리 실행
      │
      ▼
실행 결과(stdout, stderr)
      │
      ├── 문법/런타임 오류 → 에러를 LLM에 전달하여 수정
      │
      └── 실행 성공
              │
              ▼
        테스트 케이스 검증
              │
              ├── 정답 → 종료
              │
              └── 오답(Silent Wrong Answer)
                      │
                      ▼
             실패 원인과 출력 차이를 LLM에 전달
                      │
                      ▼
                  코드 수정 후 다시 실행

import subprocess
from pathlib import Path
from langchain_openai import ChatOpenAI

############################################################
# LLM
# temperature=0 : 수정 시 항상 최대한 동일한 결과를 내도록 설정
############################################################

llm = ChatOpenAI(
    model="gpt-4.1-mini",
    temperature=0
)

############################################################
# 사용자 요청
############################################################

USER_REQUEST = """
1부터 5까지 출력하는 프로그램을 작성하라.
"""

############################################################
# Planner
############################################################

planner_prompt = f"""
당신은 Coding Planner이다.

아래 요구사항을 해결하기 위한 계획만 작성하라.

요구사항
{USER_REQUEST}
"""

plan = llm.invoke(planner_prompt).content

############################################################
# 코드 생성
############################################################

def ask_code(request: str, plan: str) -> str:
    """
    최초 코드 생성
    """

    prompt = f"""
당신은 Python 전문가이다.

아래 계획을 참고하여 코드를 작성하라.

========================
계획
========================

{plan}

========================
요구사항
========================

{request}

규칙

1. 코드만 출력
2. markdown 금지
3. ``` 금지
4. 설명 금지
5. UTF-8 인코딩을 고려
6. 문자열 비교를 점검
7. 숫자 비교를 점검
8. 타입 변환(int,float,str)을 점검
9. 반환(return)과 출력(print)을 점검
10. 예외 발생 가능성을 고려
"""

    return llm.invoke(prompt).content.strip()


############################################################
# 코드 수정(Self Debugging)
############################################################

def ask_fix(
    original_code: str,
    stdout: str,
    stderr: str,
    request: str
) -> str:
    """
    실행 결과를 보고 코드 수정
    """

    prompt = f"""
당신은 Python 디버깅 전문가이다.

아래 원본 코드를 수정하라.

원본 코드는 반드시 참고한다.

```python
{original_code}
```

=========================
사용자 요구사항
=========================

{request}

=========================
실행 결과(stdout)
=========================

{stdout}

=========================
에러(stderr)
=========================

{stderr}

수정 규칙

1.
설명 금지

2.
전체 코드를 하나의 코드블록으로만 출력하지 말고
순수 코드만 출력

3.
UTF-8 저장을 고려

4.
문자열 비교를 점검

예)

if answer=="YES"

↓

answer.strip().lower()=="yes"

5.
숫자 비교를 점검

예)

"10"

↓

int("10")

6.
타입 변환 확인

7.
return / print 확인

8.
인코딩 문제 확인

9.
파일 저장 가능하도록 작성

10.
기존 기능 유지

11.
실행 가능한 완전한 코드 출력
"""

    return llm.invoke(prompt).content.strip()


############################################################
# 파일 저장
############################################################

def save_code(path: str, code: str):
    """
    코드 저장
    """

    Path(path).write_text(
        code,
        encoding="utf-8"
    )


############################################################
# subprocess 실행
############################################################

def execute(path: str):
    """
    subprocess 격리 실행

    stdout + stderr 모두 반환
    """

    result = subprocess.run(
        ["python", path],
        capture_output=True,
        text=True,
        encoding="utf-8"
    )

    output = (
        f"[stdout]\n{result.stdout}\n"
        f"[stderr]\n{result.stderr}"
    )

    return {
        "success": result.returncode == 0,
        "stdout": result.stdout,
        "stderr": result.stderr,
        "output": output
    }


############################################################
# 테스트
############################################################

EXPECTED = """1
2
3
4
5
""".strip()


############################################################
# Agent Loop
############################################################

MAX_RETRY = 5

code = ask_code(USER_REQUEST, plan)

for retry in range(MAX_RETRY):

    print("=" * 60)
    print(f"Retry : {retry+1}")
    print("=" * 60)

    ########################################################
    # solution.py 저장
    ########################################################

    save_code(
        "solution.py",
        code
    )

    ########################################################
    # 실행
    ########################################################

    result = execute("solution.py")

    ########################################################
    # Runtime Error
    ########################################################

    if not result["success"]:

        print(result["output"])

        code = ask_fix(
            original_code=code,
            stdout=result["stdout"],
            stderr=result["stderr"],
            request=USER_REQUEST
        )

        continue

    ########################################################
    # Silent Wrong Answer
    ########################################################

    actual = result["stdout"].strip()

    if actual != EXPECTED:

        print("Silent Wrong Answer")

        print("Expected")
        print(EXPECTED)

        print("Actual")
        print(actual)

        code = ask_fix(
            original_code=code,
            stdout=result["stdout"],
            stderr=f"""
실행은 성공했지만
정답이 아닙니다.

Expected

{EXPECTED}

Actual

{actual}
""",
            request=USER_REQUEST
        )

        continue

    ########################################################
    # 성공
    ########################################################

    print("Success")

    print(result["stdout"])

    break

else:

    print("최대 재시도 초과")

    save_code(
        "failed_solution.py",
        code
    )

    print("마지막 수정 코드를 failed_solution.py 에 저장했습니다.")

 

 

 

 

 

 

비즈니스 에이전트(Business Agent)

매월 반복되는 매출 보고서, 고객 분석 보고서, KPI 보고서, 운영 현황 보고서를 자동으로 생성하는 것이 목적

pandas로 계산한 facts 집계는 LLM 서술의 근거가 된다 .

실제 코드라면 Matplotlib로 저장하는 경우가 많다.

             CSV / DB
                 │
                 ▼
         Pandas 집계(Aggregation)
                 │
        ┌────────┴────────┐
        ▼                 ▼
     핵심 수치          차트 생성
        │                 │
        └────────┬────────┘
                 ▼
          LLM 서술(Narration)
                 │
                 ▼
         보고서 저장(PDF/HTML/MD)

# 원칙
# 1. 계산은 Pandas
# 2. 숫자는 facts(dict)에만 존재
# 3. LLM은 facts만 보고 서술
# 4. facts에 없는 숫자는 절대 생성 금지

def aggregate_facts(df):

    # 컬럼명 변경 대응
    alias = {
        "매출":"sales",
        "금액":"sales",
        "category":"category",
        "카테고리":"category",
        "last_sales":"last_sales"
    }

    df = df.rename(columns=alias)

    required = ["sales","category","last_sales"]

    for c in required:
        if c not in df.columns:
            raise ValueError(f"{c} 컬럼 없음")

    df["sales"] = pd.to_numeric(df["sales"], errors="coerce").fillna(0).astype(int)
    df["last_sales"] = pd.to_numeric(df["last_sales"], errors="coerce").fillna(0).astype(int)

    total = int(df["sales"].sum())
    avg = int(df["sales"].mean())
    maximum = int(df["sales"].max())
    minimum = int(df["sales"].min())

    last_total = int(df["last_sales"].sum())

    growth = 0

    if last_total != 0:
        growth = round((total-last_total)/last_total*100,2)

    category_sales = (
        df.groupby("category")["sales"]
        .sum()
        .sort_values(ascending=False)
        .to_dict()
    )

    categories = ", ".join(category_sales.keys())

    facts = {

        "총매출":total,
        "평균매출":avg,
        "최대매출":maximum,
        "최소매출":minimum,
        "전월매출":last_total,
        "증감률":growth,
        "카테고리":categories,
        "카테고리매출":category_sales

    }

    # 계산 검증
    assert facts["총매출"] >= 0
    assert isinstance(facts["총매출"],int)
    assert isinstance(facts["증감률"],(int,float))

    return facts


######################################################
# 차트 데이터
######################################################

def make_chart(df):

    chart = (
        df.groupby("category")["sales"]
        .sum()
        .sort_values(ascending=False)
    )

    chart.plot(kind="bar")

    plt.tight_layout()

    plt.savefig("sales_chart.png")

    plt.close()


######################################################
# Report
######################################################

def create_report(facts):

    prompt=f"당신은 경영 분석가이다. 반드시 facts만 근거로 보고서를 작성하라. facts에 없는 숫자를 만들지 말고 추측하지 마라. 모든 수치는 facts 그대로 사용하라. 비즈니스 관점에서 핵심성과, 위험요소, 개선제안을 작성하되 facts를 벗어나지 마라. 카테고리:{facts['카테고리']} Facts:{facts}"

    llm = ChatOpenAI(
        model="gpt-4.1-mini",
        temperature=0.3
    )

    report = llm.invoke(prompt).content

    return report


######################################################
# 캐시
######################################################

def get_cache_key(facts):

    return hashlib.sha256(
        json.dumps(
            facts,
            ensure_ascii=False,
            sort_keys=True
        ).encode()
    ).hexdigest()


######################################################
# 저장
######################################################

def save_report(report,facts):

    month = datetime.now().strftime("%Y_%m")

    folder = Path("reports")

    folder.mkdir(exist_ok=True)

    cache = folder/"cache.json"

    cache_data = {}

    if cache.exists():

        cache_data = json.loads(
            cache.read_text(
                encoding="utf-8"
            )
        )

    key = get_cache_key(facts)

    if key in cache_data:

        print("동일 데이터 → 캐시 리포트 반환")

        return cache_data[key]

    file = folder/f"report_{month}.md"

    file.write_text(
        report,
        encoding="utf-8"
    )

    cache_data[key] = report

    cache.write_text(
        json.dumps(
            cache_data,
            ensure_ascii=False,
            indent=2
        ),
        encoding="utf-8"
    )

    return report


######################################################
# Fallback
######################################################

def fallback():

    files = sorted(
        Path("reports").glob("report_*.md")
    )

    if len(files) < 2:
        return None

    return files[-2].read_text(
        encoding="utf-8"
    )


######################################################
# 알림
######################################################

def notify(msg):

    print(f"[ALERT] {msg}")


######################################################
# 실행
######################################################

try:

    facts = aggregate_facts(df)

    make_chart(df)

    report = create_report(facts)

    save_report(report,facts)

except Exception as e:

    notify(e)

    old = fallback()

    if old:

        print("직전 리포트 사용")

        print(old)

 

 

 

 

 

deployment

백엔드 개발자라면 익숙한 그 격언 — "내 PC에선 됐는데요" — 를 막기

원인은 대부분 환경 차이와 예외 미처리

deployment/
│
├── app.py                 # FastAPI
├── config.py              # .env + config.yaml
├── logging_config.py      # 로깅
├── cache.py               # Memory/File Cache
├── llm.py                 # Gemini 호출 + retry + fallback
├── report.py              # Business Agent
├── schemas.py             # Request/Response
├── requirements.txt
├── config.yaml
└── .env.example

 

 

 

requirements.txt

# ==========================
# Web Framework
# ==========================
fastapi>=0.116.0
uvicorn[standard]>=0.35.0

# ==========================
# LLM
# ==========================
langchain>=0.3.0
langchain-google-genai>=2.1.0
google-generativeai>=0.8.0

# ==========================
# Data
# ==========================
pandas>=2.3.0
numpy>=2.3.0
matplotlib>=3.10.0

# ==========================
# Config
# ==========================
python-dotenv>=1.1.0
PyYAML>=6.0

# ==========================
# Retry
# ==========================
tenacity>=9.1.0

# ==========================
# Validation
# ==========================
pydantic>=2.11.0

# ==========================
# HTTP
# ==========================
httpx>=0.28.0

# ==========================
# Optional
# ==========================
orjson>=3.11.0

.env.example

##########################################
# Gemini
##########################################

GOOGLE_API_KEY=YOUR_API_KEY

##########################################
# Logging
##########################################

LOG_LEVEL=INFO

##########################################
# Cache
##########################################

CACHE_DIR=cache

##########################################
# Report
##########################################

REPORT_DIR=reports

##########################################
# Timeout
##########################################

LLM_TIMEOUT=30

##########################################
# Retry
##########################################

MAX_RETRY=3

config.yaml

app:
  name: BusinessAgent
  version: 1.0.0

server:
  host: 0.0.0.0
  port: 8000

llm:
  model: gemini-2.5-flash
  fallback_model: gemini-2.5-flash-lite

  temperature: 0.3

  top_p: 0.9

  top_k: 40

retry:

  attempts: 3

  wait_min: 2

  wait_max: 10

logging:

  file: logs/app.log

  level: INFO

cache:

  enabled: true

  expire_hours: 24

config.py

"""
config.py

프로젝트 전체에서 사용하는 설정을 관리한다.

원칙

1.
민감정보(API Key)는 .env

2.
운영설정은 config.yaml

3.
코드에는 하드코딩하지 않는다.
"""

from pathlib import Path
import yaml
import os

from dotenv import load_dotenv

####################################################
# .env
####################################################

load_dotenv()

####################################################
# yaml
####################################################

CONFIG_PATH = Path("config.yaml")

if not CONFIG_PATH.exists():

    raise FileNotFoundError("config.yaml 없음")

with open(CONFIG_PATH, encoding="utf-8") as f:

    CONFIG = yaml.safe_load(f)

####################################################
# Config Class
####################################################

class Settings:

    ############################
    # APP
    ############################

    APP_NAME = CONFIG["app"]["name"]

    VERSION = CONFIG["app"]["version"]

    ############################
    # SERVER
    ############################

    HOST = CONFIG["server"]["host"]

    PORT = CONFIG["server"]["port"]

    ############################
    # Gemini
    ############################

    MODEL = CONFIG["llm"]["model"]

    FALLBACK_MODEL = CONFIG["llm"]["fallback_model"]

    TEMPERATURE = CONFIG["llm"]["temperature"]

    TOP_P = CONFIG["llm"]["top_p"]

    TOP_K = CONFIG["llm"]["top_k"]

    ############################
    # Retry
    ############################

    RETRY = CONFIG["retry"]["attempts"]

    WAIT_MIN = CONFIG["retry"]["wait_min"]

    WAIT_MAX = CONFIG["retry"]["wait_max"]

    ############################
    # Logging
    ############################

    LOG_LEVEL = CONFIG["logging"]["level"]

    LOG_FILE = CONFIG["logging"]["file"]

    ############################
    # Cache
    ############################

    CACHE_ENABLE = CONFIG["cache"]["enabled"]

    CACHE_HOURS = CONFIG["cache"]["expire_hours"]

    ############################
    # ENV
    ############################

    GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")

    CACHE_DIR = os.getenv("CACHE_DIR", "cache")

    REPORT_DIR = os.getenv("REPORT_DIR", "reports")

    LLM_TIMEOUT = int(os.getenv("LLM_TIMEOUT", 30))

####################################################
# Singleton
####################################################

settings = Settings()

logging_config.py

"""
logging_config.py

운영에서는 print() 대신 logging을 사용한다.

이유

1.
로그 레벨 구분

DEBUG

INFO

WARNING

ERROR

CRITICAL

2.
콘솔 + 파일 동시 기록

3.
운영 중 장애 분석 가능

4.
모든 모듈에서 동일 Logger 사용
"""

import logging

from pathlib import Path

from config import settings

######################################################
# Log Folder
######################################################

log_file = Path(settings.LOG_FILE)

log_file.parent.mkdir(
    exist_ok=True,
    parents=True
)

######################################################
# Logger
######################################################

logger = logging.getLogger("BusinessAgent")

logger.setLevel(settings.LOG_LEVEL)

######################################################
# Formatter
######################################################

formatter = logging.Formatter(

    "%(asctime)s | %(levelname)s | %(filename)s:%(lineno)d | %(message)s"

)

######################################################
# Console
######################################################

console = logging.StreamHandler()

console.setFormatter(formatter)

######################################################
# File
######################################################

file = logging.FileHandler(

    settings.LOG_FILE,

    encoding="utf-8"

)

file.setFormatter(formatter)

######################################################
# 중복 Handler 방지
######################################################

if not logger.handlers:

    logger.addHandler(console)

    logger.addHandler(file)

######################################################
# 사용 예시
######################################################

"""
logger.debug("Debug")

logger.info("Start")

logger.warning("Warning")

logger.error("Network Error")

logger.critical("Fatal Error")
"""

cache.py

"""
cache.py

Business Agent Cache

역할
1. 질문 정규화(Normalize)
2. SHA256 Cache Key 생성
3. Memory Cache (빠른 조회)
4. File Cache (영속 저장)
5. Cache 우선 조회 후 LLM 호출
6. Cache 만료(TTL) 지원

원칙
- 동일한 질문은 LLM을 다시 호출하지 않는다.
- 질문을 정규화하여 Cache Hit를 높인다.
- Memory -> File 순으로 조회한다.
- JSON으로 영속 저장한다.
"""

from __future__ import annotations

import hashlib
import json
import re
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any

from config import settings
from logging_config import logger


class Cache:

    def __init__(self):

        self.memory: dict[str, dict] = {}

        self.cache_dir = Path(settings.CACHE_DIR)

        self.cache_dir.mkdir(
            exist_ok=True,
            parents=True
        )

        self.cache_file = self.cache_dir / "cache.json"

        self.expire = timedelta(
            hours=settings.CACHE_HOURS
        )

        self.file_cache = self._load_file_cache()

    #######################################################
    # 질문 정규화
    #######################################################

    def normalize(self, question: str) -> str:
        """
        Cache Hit를 높이기 위해 질문을 정규화한다.

        예)

        "   매출 알려줘!! "

        ↓

        "매출 알려줘"
        """

        question = question.strip()

        question = question.lower()

        question = re.sub(r"\s+", " ", question)

        return question

    #######################################################
    # Cache Key
    #######################################################

    def make_key(self, question: str) -> str:

        normalized = self.normalize(question)

        return hashlib.sha256(
            normalized.encode("utf-8")
        ).hexdigest()

    #######################################################
    # 파일 읽기
    #######################################################

    def _load_file_cache(self) -> dict:

        if not self.cache_file.exists():

            logger.info("Cache File 생성")

            return {}

        try:

            return json.loads(

                self.cache_file.read_text(
                    encoding="utf-8"
                )

            )

        except Exception as e:

            logger.warning(f"Cache Load 실패 : {e}")

            return {}

    #######################################################
    # 파일 저장
    #######################################################

    def _save_file_cache(self):

        try:

            self.cache_file.write_text(

                json.dumps(

                    self.file_cache,

                    ensure_ascii=False,

                    indent=2

                ),

                encoding="utf-8"

            )

        except Exception as e:

            logger.error(f"Cache Save 실패 : {e}")

    #######################################################
    # 만료 검사
    #######################################################

    def _expired(self, created_at: str) -> bool:

        created = datetime.fromisoformat(created_at)

        return datetime.now() > created + self.expire

    #######################################################
    # 조회
    #######################################################

    def get(self, question: str):

        key = self.make_key(question)

        ###################################################
        # Memory Cache
        ###################################################

        if key in self.memory:

            cache = self.memory[key]

            if not self._expired(cache["created_at"]):

                logger.info("Memory Cache Hit")

                return cache["response"]

            logger.info("Memory Cache Expired")

            del self.memory[key]

        ###################################################
        # File Cache
        ###################################################

        if key in self.file_cache:

            cache = self.file_cache[key]

            if not self._expired(cache["created_at"]):

                logger.info("File Cache Hit")

                self.memory[key] = cache

                return cache["response"]

            logger.info("File Cache Expired")

            del self.file_cache[key]

            self._save_file_cache()

        ###################################################
        # Miss
        ###################################################

        logger.info("Cache Miss")

        return None

    #######################################################
    # 저장
    #######################################################

    def set(
        self,
        question: str,
        response: Any
    ):

        key = self.make_key(question)

        data = {

            "question": self.normalize(question),

            "response": response,

            "created_at": datetime.now().isoformat()

        }

        ###################################################
        # Memory
        ###################################################

        self.memory[key] = data

        ###################################################
        # File
        ###################################################

        self.file_cache[key] = data

        self._save_file_cache()

        logger.info("Cache Saved")

    #######################################################
    # 삭제
    #######################################################

    def delete(self, question: str):

        key = self.make_key(question)

        self.memory.pop(key, None)

        self.file_cache.pop(key, None)

        self._save_file_cache()

        logger.info("Cache Deleted")

    #######################################################
    # 전체 삭제
    #######################################################

    def clear(self):

        self.memory.clear()

        self.file_cache.clear()

        self._save_file_cache()

        logger.warning("Cache Cleared")

    #######################################################
    # 통계
    #######################################################

    def stats(self):

        return {

            "memory_count": len(self.memory),

            "file_count": len(self.file_cache),

            "expire_hours": settings.CACHE_HOURS

        }


############################################################
# Singleton
############################################################

cache = Cache()


############################################################
# Example
############################################################
"""
question = "매출 알려줘"

cached = cache.get(question)

if cached:

    print("Cache 사용")

    print(cached)

else:

    response = "이번 달 총매출은 1,250만원입니다."

    cache.set(question, response)

    print(response)
"""

llm.py

# llm.py

class LLMClient:

    def __init__(self):

        logger.info("LLM Initialize")

        self.primary=ChatGoogleGenerativeAI(
            model=settings.MODEL,
            google_api_key=settings.GOOGLE_API_KEY,
            temperature=settings.TEMPERATURE,
            top_p=settings.TOP_P,
            top_k=settings.TOP_K,
            timeout=settings.LLM_TIMEOUT
        )

        self.fallback=ChatGoogleGenerativeAI(
            model=settings.FALLBACK_MODEL,
            google_api_key=settings.GOOGLE_API_KEY,
            temperature=settings.TEMPERATURE,
            timeout=settings.LLM_TIMEOUT
        )

    ##################################################
    # Retry
    # wait_random_exponential = exponential backoff+jitter
    # retry_if_exception_type = 지정 예외만 재시도
    ##################################################

    @retry(
        stop=stop_after_attempt(settings.RETRY),
        wait=wait_random_exponential(multiplier=1,max=settings.WAIT_MAX),
        retry=retry_if_exception_type(Exception),
        reraise=True
    )
    def _invoke(self,llm,prompt):

        logger.info("LLM Request")

        response=llm.invoke(prompt)

        if not response.content:
            raise RuntimeError("Empty Response")

        return response.content

    ##################################################
    # Robust Invoke
    ##################################################

    def robust_invoke(self,question):

        question=" ".join(question.strip().lower().split())

        cached=cache.get(question)

        if cached:

            logger.info("Cache Hit")

            return cached

        prompt=f"당신은 비즈니스 AI이다. 반드시 질문에만 답하고 추측하지 마라. 사실이 부족하면 부족하다고 말하라. 질문:{question}"

        ##################################################
        # Primary
        ##################################################

        try:

            logger.info(f"Primary Model : {settings.MODEL}")

            answer=self._invoke(self.primary,prompt)

            cache.set(question,answer)

            return answer

        ##################################################
        # Rate Limit / Timeout / Network
        ##################################################

        except Exception as e:

            logger.warning(f"Primary Fail : {e}")

        ##################################################
        # Fallback
        ##################################################

        try:

            logger.info(f"Fallback Model : {settings.FALLBACK_MODEL}")

            answer=self._invoke(self.fallback,prompt)

            cache.set(question,answer)

            return answer

        except Exception as e:

            logger.error(f"Fallback Fail : {e}")

            raise RuntimeError("모든 LLM 호출 실패")

    ##################################################
    # Report Prompt
    ##################################################

    def report(self,facts):

        categories=", ".join(facts["카테고리매출"].keys())

        prompt=f"너는 경영 분석가이다. facts만 근거로 보고서를 작성하라. facts에 없는 숫자를 만들지 마라. 추측하지 마라. 확정된 수치만 사용하라. 수치의 의미를 설명하고 위험요소와 개선제안을 작성하되 facts를 벗어나지 마라. 카테고리:{categories} Facts:{facts}"

        return self.robust_invoke(prompt)

##################################################
# Singleton
##################################################

llm=LLMClient()

##################################################
# Example
##################################################

"""
answer=llm.robust_invoke("이번달 매출 알려줘")

facts={
    "총매출":1200000,
    "평균매출":300000,
    "증감률":8.2,
    "카테고리매출":{
        "전자":500000,
        "식품":400000,
        "의류":300000
    }
}

report=llm.report(facts)

print(report)
"""

schemas.py

# schemas.py

"""
FastAPI Request / Response Schema

역할
1. 요청 데이터 검증
2. 응답 형식 통일
3. API 계약(API Contract) 정의
4. Swagger(OpenAPI) 자동 문서 생성

원칙
- 모든 API는 Request/Response Schema를 사용한다.
- dict를 직접 반환하지 않는다.
"""

####################################################
# Report Request
####################################################

class ReportRequest(BaseModel):

    question:str=Field(
        ...,
        description="사용자 질문",
        example="이번 달 매출 보고서를 작성해줘"
    )

    data:list[dict]=Field(
        ...,
        description="집계할 데이터"
    )

####################################################
# Chat Request
####################################################

class ChatRequest(BaseModel):

    question:str=Field(
        ...,
        min_length=1,
        max_length=500,
        description="사용자 질문"
    )

####################################################
# Fact Model
####################################################

class Facts(BaseModel):

    총매출:int

    평균매출:int

    최대매출:int

    최소매출:int

    전월매출:int

    증감률:float

    카테고리:str

    카테고리매출:dict[str,int]

####################################################
# Report Response
####################################################

class ReportResponse(BaseModel):

    success:bool=True

    report:str

    facts:Facts

    chart_path:str|None=None

    cached:bool=False

    created_at:datetime=Field(
        default_factory=datetime.now
    )

####################################################
# Chat Response
####################################################

class ChatResponse(BaseModel):

    success:bool=True

    answer:str

    cached:bool=False

    model:str

    created_at:datetime=Field(
        default_factory=datetime.now
    )

####################################################
# Error Response
####################################################

class ErrorResponse(BaseModel):

    success:bool=False

    error:str

    detail:str|None=None

    created_at:datetime=Field(
        default_factory=datetime.now
    )

####################################################
# Health Check
####################################################

class HealthResponse(BaseModel):

    status:str="ok"

    app:str=settings.APP_NAME

    version:str=settings.VERSION

    model:str=settings.MODEL

    cache:bool=settings.CACHE_ENABLE

    timestamp:datetime=Field(
        default_factory=datetime.now
    )

####################################################
# Cache Response
####################################################

class CacheResponse(BaseModel):

    memory_count:int

    file_count:int

    expire_hours:int

####################################################
# Example
####################################################

"""
POST /chat

{
  "question":"매출 알려줘"
}

↓

{
  "success":true,
  "answer":"이번 달 총매출은 ...",
  "cached":false,
  "model":"gemini-2.5-flash",
  "created_at":"2026-07-21T10:00:00"
}

-----------------------------------------

POST /report

{
  "question":"보고서 작성",
  "data":[...]
}

↓

{
  "success":true,
  "report":"...",
  "facts":{
      ...
  },
  "chart_path":"reports/chart.png",
  "cached":false
}
"""

report.py

# report.py

"""
Business Agent Report Module

역할
1. 데이터 검증
2. Pandas 기반 Facts 생성
3. 차트 데이터 생성
4. LLM 서술 생성
5. 리포트 저장

원칙
- 숫자 계산은 Python/Pandas가 담당
- LLM은 Facts를 기반으로 서술만 담당
- Facts 외 숫자 생성 금지
"""

# import 생략


##################################################
# Facts 집계
##################################################

def aggregate_facts(df):

    logger.info("Start aggregate facts")

    # 컬럼 변경 대응
    alias = {
        "매출":"sales",
        "판매금액":"sales",
        "카테고리":"category",
        "분류":"category",
        "전월매출":"last_sales"
    }

    df = df.rename(columns=alias)


    required=[
        "sales",
        "category",
        "last_sales"
    ]

    for col in required:

        if col not in df.columns:

            raise ValueError(
                f"필수 컬럼 없음 : {col}"
            )


    # 타입 안정화
    df["sales"] = pd.to_numeric(
        df["sales"],
        errors="coerce"
    ).fillna(0).astype(int)


    df["last_sales"] = pd.to_numeric(
        df["last_sales"],
        errors="coerce"
    ).fillna(0).astype(int)



    total=int(
        df["sales"].sum()
    )


    previous=int(
        df["last_sales"].sum()
    )


    growth=0

    if previous:

        growth=round(
            ((total-previous)/previous)*100,
            2
        )


    category_sales=(
        df.groupby("category")
        ["sales"]
        .sum()
        .astype(int)
        .sort_values(
            ascending=False
        )
        .to_dict()
    )


    facts={

        "총매출":total,

        "평균매출":int(
            df["sales"].mean()
        ),

        "최대매출":int(
            df["sales"].max()
        ),

        "최소매출":int(
            df["sales"].min()
        ),

        "전월매출":previous,

        "증감률":growth,

        "카테고리":", ".join(
            category_sales.keys()
        ),

        "카테고리매출":category_sales

    }


    # 계산 결과 검증

    assert isinstance(
        facts["총매출"],
        int
    )

    assert isinstance(
        facts["증감률"],
        float|int
    )


    logger.info(
        f"Facts 생성 완료 : {facts}"
    )


    return facts



##################################################
# Chart 생성
##################################################

def create_chart(df):

    logger.info(
        "Create chart"
    )


    chart=df.groupby(
        "category"
    )["sales"].sum()


    path=Path(
        settings.REPORT_DIR
    )

    path.mkdir(
        exist_ok=True
    )


    file=path/"category_sales.png"


    chart.plot(
        kind="bar"
    )


    plt.tight_layout()


    plt.savefig(
        file
    )


    plt.close()


    return str(file)



##################################################
# Report 생성
##################################################

def generate_report(facts):


    # 중요:
    # LLM에게 원본 데이터가 아닌
    # 검증 완료된 facts만 전달


    categories=", ".join(
        facts["카테고리매출"].keys()
    )


    prompt=f"너는 기업 데이터 분석가다. 아래 facts만 근거로 월간 보고서를 작성하라. facts에 없는 숫자를 만들지 마라. 추측하지 마라. 모든 수치는 facts 값을 그대로 사용하라. 분석에는 주요성과, 감소요인, 개선제안을 포함하라. 제안은 facts 기반으로 작성하라. 카테고리:{categories} facts:{facts}"


    report=llm.robust_invoke(
        prompt
    )


    return report



##################################################
# Report 저장
##################################################

def save_report(report,facts):


    report_dir=Path(
        settings.REPORT_DIR
    )


    report_dir.mkdir(
        exist_ok=True
    )


    month=datetime.now().strftime(
        "%Y_%m"
    )


    file=report_dir / (
        f"business_report_{month}.md"
    )


    # 기존 파일 보호
    if file.exists():

        logger.warning(
            "동일 월 리포트 존재"
        )


        backup=report_dir / (
            f"business_report_{month}_backup.md"
        )

        shutil.copy(
            file,
            backup
        )


    file.write_text(
        report,
        encoding="utf-8"
    )


    logger.info(
        f"Report saved : {file}"
    )


    return str(file)



##################################################
# 전체 실행
##################################################

def run_report(df):


    try:


        # 1. 숫자 계산
        facts=aggregate_facts(
            df
        )


        # 2. 차트
        chart=create_chart(
            df
        )


        # 3. LLM 서술
        report=generate_report(
            facts
        )


        # 4. 저장
        path=save_report(
            report,
            facts
        )


        return {

            "facts":facts,

            "report":report,

            "chart":chart,

            "path":path,

            "success":True

        }


    except Exception as e:


        logger.error(
            f"Report 실패 : {e}"
        )


        old=fallback()


        if old:

            return {

                "success":False,

                "report":old,

                "fallback":True

            }


        raise e



##################################################
# Fallback
##################################################

def fallback():

    files=sorted(
        Path(
            settings.REPORT_DIR
        ).glob(
            "business_report_*.md"
        )
    )


    if len(files)==0:

        return None


    return files[-1].read_text(
        encoding="utf-8"
    )

app.py

# app.py

"""
FastAPI Application

역할
1. HTTP API 제공
2. Request 검증
3. Business Agent 실행
4. Response Schema 반환
5. 운영용 Health Check 제공

운영 원칙
- print 사용 금지
- 모든 요청/오류 logging
- LLM은 import 시 1회 초기화된 Singleton 사용
- API 계약은 schemas.py 기준
"""

# import 생략


##################################################
# FastAPI 생성
##################################################

app = FastAPI(

    title=settings.APP_NAME,

    version=settings.VERSION,

    description="Business Agent API"

)


##################################################
# Startup
##################################################

@app.on_event("startup")
async def startup():

    logger.info(
        "Application Started"
    )

    logger.info(
        f"Model : {settings.MODEL}"
    )

    logger.info(
        f"Cache : {settings.CACHE_ENABLE}"
    )


##################################################
# Health Check
##################################################

@app.get(
    "/health",
    response_model=HealthResponse
)
async def health():

    return HealthResponse()



##################################################
# 일반 LLM 요청
##################################################

@app.post(
    "/chat",
    response_model=ChatResponse
)
async def chat(
    request:ChatRequest
):

    try:

        logger.info(
            "Chat Request"
        )


        cached = cache.get(
            request.question
        )


        if cached:

            return ChatResponse(

                answer=cached,

                cached=True,

                model=settings.MODEL

            )


        answer = llm.robust_invoke(
            request.question
        )


        cache.set(
            request.question,
            answer
        )


        return ChatResponse(

            answer=answer,

            cached=False,

            model=settings.MODEL

        )


    except Exception as e:


        logger.error(
            f"Chat Error : {e}"
        )


        raise HTTPException(

            status_code=500,

            detail="LLM 처리 실패"

        )



##################################################
# Business Report
##################################################

@app.post(
    "/report",
    response_model=ReportResponse
)
async def report(
    request:ReportRequest
):

    try:

        logger.info(
            "Report Request"
        )


        df=pd.DataFrame(
            request.data
        )


        result=run_report(
            df
        )


        return ReportResponse(

            report=result["report"],

            facts=result["facts"],

            chart_path=result.get(
                "chart"
            ),

            cached=result.get(
                "cached",
                False
            )

        )


    except Exception as e:


        logger.error(
            f"Report Error : {e}"
        )


        raise HTTPException(

            status_code=500,

            detail="Report 생성 실패"

        )



##################################################
# Cache 상태 확인
##################################################

@app.get(
    "/cache",
    response_model=CacheResponse
)
async def cache_status():


    result=cache.stats()


    return CacheResponse(

        memory_count=result[
            "memory_count"
        ],

        file_count=result[
            "file_count"
        ],

        expire_hours=result[
            "expire_hours"
        ]

    )



##################################################
# Error Handler
##################################################

@app.exception_handler(Exception)
async def global_exception_handler(
    request,
    exc
):

    logger.error(
        f"Unhandled Error : {exc}"
    )


    return JSONResponse(

        status_code=500,

        content={

            "success":False,

            "error":"Internal Server Error"

        }

    )



##################################################
# Local 실행
##################################################

"""
실행

uvicorn app:app --host 0.0.0.0 --port 8000


API

GET

/health


POST

/chat

{
    "question":"이번달 매출 알려줘"
}


POST

/report

{
    "question":"월간 보고서 작성",

    "data":[

        {
            "category":"전자",
            "sales":500000,
            "last_sales":400000
        },

        {
            "category":"식품",
            "sales":300000,
            "last_sales":350000
        }

    ]
}


운영 흐름

Client
 |
 | HTTP
 ▼
FastAPI
 |
 ▼
Schema Validation
 |
 ▼
Business Agent
 |
 ├── Cache 확인
 |
 ├── Pandas Facts 생성
 |
 ├── Chart 생성
 |
 └── LLM Report 생성

 |
 ▼

Response
"""