[17편] Part 2 GPT 컨텍스트 엔지니어링

 

📘 Part 2 — GPT 컨텍스트 엔지니어링 전체 코드 + 설명

이번 Part 2에서는
**GPT를 “많이 써도 느려지지 않고, 비용과 맥락을 동시에 제어할 수 있는 구조”**로 만드는
컨텍스트 엔지니어링 전체 설계를 정리합니다.

핵심 목표는 다음 한 문장입니다.

긴 대화에서도 응답 속도는 유지하면서, 중요한 맥락만 정확히 전달하는 GPT 구조

---

✅ 포함 파일 목록

• requirements.txt
• token_utils.py
• context_builder.py
• memory_service.py
• gpt_service.py
• chats.py
• chat.py
• main.py

---

1️⃣ requirements.txt

fastapi
uvicorn
python-dotenv
openai
httpx
tiktoken
mysql-connector-python

✅ 왜 추가했나요?

• tiktoken
  → GPT 토큰 계산을 위한 필수 라이브러리입니다.
• mysql-connector-python
  → MySQL 기반 대화 / 요약 데이터 저장을 위해 필요합니다.

✅ 장점

• 토큰 기반 비용 통제가 가능해집니다.
• 장기 대화 데이터를 안정적으로 저장할 수 있습니다.

---

2️⃣ token_utils.py

import tiktoken

def count_tokens(text: str, model: str = "gpt-4o-mini") -> int:
    try:
        encoding = tiktoken.encoding_for_model(model)
    except KeyError:
        encoding = tiktoken.get_encoding("cl100k_base")
    return len(encoding.encode(text))

SYSTEM_PROMPT = (
    "당신은 패션 코디 추천 AI 'Aura'입니다. "
    "날씨와 취향에 맞는 옷을 추천하세요. "
    "답변은 JSON만 출력합니다. "
    "코디 추천 요청이면 top/bottom/shoes/outer/reason을 채우고, "
    "코디와 무관한 대화면 message 필드만 채우세요. "
    'JSON 예시: {"top":"","bottom":"","shoes":"","outer":"","reason":"","message":""}'
)

SYSTEM_PROMPT_TOKENS = count_tokens(SYSTEM_PROMPT)

✅ 문법 설명

• try / except
  → 모델에 맞는 인코딩이 없을 경우 기본 인코딩으로 fallback 처리합니다.
• SYSTEM_PROMPT_TOKENS
  → 시스템 프롬프트 토큰을 미리 계산하여 성능을 개선합니다.

✅ 장점

• 요청마다 토큰 계산 반복을 방지합니다.
• 응답 속도가 안정적으로 유지됩니다.
• 비용 예측이 가능해집니다.

---

3️⃣ context_builder.py

import logging
import os

from services.token_utils import (
    SYSTEM_PROMPT,
    SYSTEM_PROMPT_TOKENS,
    count_tokens,
)

MAX_TOKENS = int(os.getenv("MAX_CONTEXT_TOKENS", 3000))
MAX_MESSAGES = int(os.getenv("MAX_RECENT_MESSAGES", 30))

def build_smart_context(
    history: list[dict],
    current_message: str,
    summary: str | None = None,
) -> list[dict]:
    recent_history = history[-MAX_MESSAGES:]

    final_messages: list[dict] = []
    current_total_tokens = SYSTEM_PROMPT_TOKENS

    base_content = SYSTEM_PROMPT
    if summary:
        summary_text = f"\n[이전 대화 요약]: {summary}"
        base_content += summary_text
        current_total_tokens += count_tokens(summary_text)

    final_messages.append({"role": "system", "content": base_content})

    current_total_tokens += count_tokens(current_message)

    trimmed_history: list[dict] = []
    for msg in reversed(recent_history):
        msg_tokens = count_tokens(msg.get("content", ""))
        if current_total_tokens + msg_tokens > MAX_TOKENS:
            break
        trimmed_history.insert(0, msg)
        current_total_tokens += msg_tokens

    final_messages.extend(trimmed_history)
    final_messages.append({"role": "user", "content": current_message})

    logging.info(
        "[Context Engine] tokens=%s, history_used=%s/%s, summary_used=%s",
        current_total_tokens,
        len(trimmed_history),
        len(recent_history),
        bool(summary),
    )

    return final_messages

✅ 설계 이유

• MAX_MESSAGES
  → 최근 대화만 우선 사용합니다.
• MAX_TOKENS
  → 토큰 초과 시 자동으로 과거 메시지를 제거합니다.
• summary
  → 오래된 대화를 요약으로 보존합니다.

✅ 장점

• 대화가 길어져도 응답 속도가 일정합니다.
• 비용 폭발을 방지합니다.
• 핵심 맥락은 유지됩니다.

---

4️⃣ memory_service.py

import os

from services.message_service import update_session_summary
from services.summary_service import summarize_messages

SUMMARY_TRIGGER = int(os.getenv("SUMMARY_TRIGGER_COUNT", 40))

async def check_and_update_summary(session_id: int, history: list[dict]):
    if len(history) >= SUMMARY_TRIGGER:
        content_list = [m.get("content", "") for m in history]
        new_summary = await summarize_messages(content_list)
        update_session_summary(session_id, new_summary)
        return new_summary
    return None

✅ 설계 이유

• 대화가 일정 길이를 넘으면 요약을 생성합니다.
• 요약을 DB에 저장해 장기 기억을 유지합니다.

✅ 장점

기억은 유지하고, 비용은 절감합니다.

---

5️⃣ gpt_service.py

import logging
import os

from dotenv import load_dotenv
from openai import AsyncOpenAI

from services.context_builder import build_smart_context
from services.memory_service import check_and_update_summary

load_dotenv()

_client: AsyncOpenAI | None = None

def _get_client() -> AsyncOpenAI:
    global _client
    if _client is None:
        _client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    return _client

async def get_aura_response(
    session_id: int,
    user_input: str,
    history: list[dict],
    current_summary: str | None,
) -> str:
    api_key = os.getenv("OPENAI_API_KEY")
    if not api_key:
        return '{"message":"GPT API 키가 설정되지 않았습니다."}'

    messages = build_smart_context(history, user_input, current_summary)

    try:
        client = _get_client()
        response = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            temperature=0.6,
            timeout=20,
            response_format={"type": "json_object"},
        )
        content = response.choices[0].message.content
        reply = content or '{"message":"응답을 생성하지 못했습니다."}'
    except Exception:
        logging.exception("GPT Error")
        reply = (
            '{"message":"GPT 응답 생성 중 문제가 발생했습니다. 잠시 후 다시 시도해주세요."}'
        )

    await check_and_update_summary(session_id, history)
    return reply

✅ 문법 포인트

• async / await
  → 서버 블로킹을 방지합니다.
• response_format={"type":"json_object"}
  → GPT 응답을 JSON으로 강제합니다.

✅ 장점

• 프론트엔드에서 JSON.parse가 안전합니다.
• 카드형 UI, 자동 렌더링이 가능합니다.

---

6️⃣ chats.py

from fastapi import APIRouter, Header, HTTPException

from schemas.chat import ChatRequest, ChatResponse
from services.gpt_service import get_aura_response
from services.message_service import (
    get_messages_by_session,
    get_session,
    get_session_summary,
)
from services.weather_service import get_weather

router = APIRouter(prefix="/chats")

@router.post("/start", response_model=ChatResponse)
async def start_chat(
    body: ChatRequest,
    x_user_id: str | None = Header(default=None),
):
    if not x_user_id:
        raise HTTPException(status_code=401, detail="Unauthorized")

    session = get_session(body.session_id)
    if not session or session["user_id"] != x_user_id:
        raise HTTPException(status_code=404, detail="Session not found")

    weather = await get_weather(body.lat, body.lon)
    full_context = (
        "날짜: 오늘\n"
        "위치: 사용자 위치\n"
        f"날씨: {weather}\n"
        f"{body.context}"
    )

    history = [
        {"role": item["role"], "content": item["content"]}
        for item in get_messages_by_session(body.session_id)
        if item.get("role") in {"user", "assistant"}
    ]
    current_summary = get_session_summary(body.session_id)
    current_message = f"[상황]\n{full_context}\n\n[요청]\n{body.message}"

    reply = await get_aura_response(
        body.session_id,
        current_message,
        history,
        current_summary,
    )
    return ChatResponse(reply=reply)

✅ 장점

• 세션 기반 히스토리 연결
• 날씨 + 요약 + 사용자 요청을 하나의 맥락으로 구성

---

7️⃣ chat.py

from pydantic import BaseModel

class ChatRequest(BaseModel):
    message: str
    context: str
    session_id: int
    lat: float = 37.5665
    lon: float = 126.9780

class ChatResponse(BaseModel):
    reply: str

✅ 장점

• 타입 안정성 확보
• Swagger 자동 문서화 지원

---

8️⃣ main.py

import logging
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from dotenv import load_dotenv

from routers.health import router as health_router
from routers.users import router as users_router
from routers.chats import router as chats_router
from routers.summaries import router as summaries_router
from routers.sessions import router as sessions_router
from routers.messages import router as messages_router
from routers.locations import router as locations_router

logging.basicConfig(level=logging.INFO)
load_dotenv()
app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(health_router)
app.include_router(users_router)
app.include_router(chats_router)
app.include_router(summaries_router)
app.include_router(sessions_router)
app.include_router(messages_router)
app.include_router(locations_router)

✅ 장점

• 컨텍스트 및 토큰 로그 확인 가능
• 실서비스 구조에 바로 적용 가능

---

✅ Part 2 핵심 정리

• GPT 컨텍스트를 토큰 기준으로 제어
• 오래된 대화는 요약으로 기억
• 응답은 항상 JSON으로 강제
• 비용 / 속도 / 맥락을 동시에 관리하는 구조

---