[ SK 네트웍스 Family AI 캠프 수업 내용 복습 ] 벡터 데이터 베이스( 벡터 DB의 이해, 벡터 DB를 활용한 검색 )-2 2025-05-14

프롬프트 엔지니어링 응용 (LangChain)

MoE (Mixture of Experts)

**Mixture of Experts(MoE)**는 대규모 언어 모델(LLM)의 연산 효율을 극대화하기 위한 아키텍처 설계 방식 중 하나입니다.

1. 기존 LLM과의 차이점

• 기존 LLM (예: GPT, PaLM 등)
  • 모든 입력에 대해 전체 파라미터(모델의 모든 층과 유닛)를 항상 사용합니다.
  • 파라미터 수가 많을수록 연산량이 커지고, 학습 및 추론 비용이 높아집니다.
• MoE 기반 LLM (예: Mixtral 8x7B)
  • 모델 내 여러 개의 전문가 네트워크(Experts)가 존재합니다.
  • 입력에 따라 그 중 일부 전문가만 선택되어 활성화됩니다.
  • 예: 8개의 전문가 중 2개만 활성화되면 7B x 2 = 14B 파라미터만 사용 → 전체 56B 중 일부만 사용.

2. MoE의 핵심 원리

• 전문가(Expert): 여러 개의 하위 모델(일반적으로 MLP 계층)로 구성됩니다.
• 라우터(Router): 입력 토큰을 받아 어떤 전문가를 사용할지 결정합니다.
• Top-k Routing: 가장 관련 있는 상위 k개의 전문가를 선택해 계산을 수행합니다. (예: Top-2)

3. 장점

• 효율성 향상: 전체 모델 크기는 크지만, 연산 시 사용하는 파라미터 수는 일부만 사용하므로 연산량 감소.
• 추론 속도 증가: 빠른 처리 가능.
• 성능 유지: 전체 모델 용량은 그대로이기 때문에 성능은 높게 유지 가능.

4. 비교 예시

모델	구조 유형	파라미터 사용 방식
GPT	거대 단일 모델	항상 전체 파라미터 사용
Mixtral 8x7B	MoE 구조	일부 전문가만 선택적으로 사용

 

---

Step-by-Step Reasoning

Step-by-Step Reasoning은 복잡한 문제나 질문에 대해 단계를 나누어 순차적으로 추론하는 방법을 말합니다.
LLM이 복잡한 질문에 대해 더 정확하게 대답하도록 도와주는 대표적인 프롬프트 엔지니어링 기법 중 하나입니다.

---

1. 개념 설명

• 문제를 한 번에 해결하려 하지 않고,
• 문제를 구성하는 하위 단계들로 나누어 각 단계별로 추론을 수행합니다.

이 방식은 수학 문제, 논리 문제, 가격 계산 등과 같이 논리적이고 절차적인 사고가 필요한 상황에서 특히 효과적입니다.

---

2. 예시

문제:
"OO 마트에서 사과 2개, 오렌지 2개, 바나나 4개를 샀다. 총액은?"

Step-by-Step Reasoning 방식의 출력:

1. 데이터 조회:
   • 사과 1개의 가격은 1,000원
   • 오렌지 1개의 가격은 1,500원
   • 바나나 1개의 가격은 500원
2. 할인 적용:
   • 현재 오렌지는 1+1 행사 중이므로 2개 가격은 1개 가격으로 계산됨
   • 바나나는 3개 이상 구매 시 10% 할인
3. 계산:
   • 사과 2개: 1,000 x 2 = 2,000원
   • 오렌지 2개: 1,500 x 1 = 1,500원
   • 바나나 4개: 500 x 4 = 2,000원 → 10% 할인 적용: 2,000 x 0.9 = 1,800원
4. 검증 및 합산:
   • 총합: 2,000 + 1,500 + 1,800 = 5,300원

최종 출력:
"총액은 5,300원입니다."

---

3. 활용 맥락

• **Chain of Thought (CoT)**와 유사한 방식이며, 실제 CoT 프롬프트에서 사용됨.
• LangChain, AutoGPT, Agent 기반 시스템에서도 내부 reasoning 루틴으로 사용됨.
• 실세계 문제 해결, 금융 계산, 재고 관리, 조건 분기 처리 등 다양한 응용 가능.

 

---

# 과일 가격 정보 딕셔너리 정의
FRUIT_PRICE = {
    "사과": {"price_per_kg": 7000, "units_per_kg": 4},    # 1개당 가격: 7000 ÷ 4 = 1750원
    "오렌지": {"price_per_kg": 8000, "units_per_kg": 5},  # 1개당 가격: 8000 ÷ 5 = 1600원
    "바나나": {"price_per_kg": 4500, "units_per_kg": 6}   # 1개당 가격: 4500 ÷ 6 = 750원
}

# 할인 기준 및 할인율 설정
DISCOUNT_THRESHOLD = 10000  # 총액이 10,000원 이상일 경우
DISCOUNT_RATE = 0.1         # 10% 할인 적용

# Pydantic을 이용한 요청 데이터 모델 정의
from pydantic import BaseModel, Field
from typing import Dict

class PurchaseRequest(BaseModel):
    # items: 과일 이름(str)과 수량(int)을 가지는 딕셔너리
    items: Dict[str, int] = Field(..., description='과일 이름과 수량(예: {"사과": 3, "오렌지": 4})')

# 예시 요청 생성 (사과 4개, 오렌지 2개, 바나나 4개 구매)
data = PurchaseRequest(items={"사과": 4, "오렌지": 2, "바나나": 4})

# 요청 객체 출력
print(data)

# 딕셔너리 형태로 추출된 items 출력
print(data.items)

코드 설명

• FRUIT_PRICE: 과일의 킬로그램당 가격과 1kg당 개수 정보를 갖고 있는 딕셔너리입니다. 이를 통해 개당 가격을 계산할 수 있습니다.
• DISCOUNT_THRESHOLD, DISCOUNT_RATE: 할인을 적용할 기준을 정의합니다. 총액이 일정 기준 이상이면 일정 비율만큼 할인됩니다.
• PurchaseRequest: pydantic.BaseModel을 상속받아 사용자의 구매 요청이 구조적으로 올바른지 타입 검증을 수행합니다.
• data = PurchaseRequest(...): 예시 데이터를 검증하고 모델 인스턴스로 변환합니다.
• print(data.items): 검증된 요청에서 과일과 수량의 딕셔너리를 확인할 수 있습니다.

 

---

# LangChain 관련 패키지를 설치 (조용히 설치)
!pip install langchain_community -q
!pip install --upgrade langchain langchain-community -q

# OpenAI GPT 모델을 LangChain과 연동
from langchain.chat_models import ChatOpenAI
import os

# OpenAI API 키를 환경 변수로 설정 (보안을 위해 외부에 노출하지 않음)
os.environ["OPENAI_API_KEY"] = 'private api key'

# GPT-3.5 Turbo 모델을 사용하는 LangChain LLM 객체 생성
llm = ChatOpenAI(model='gpt-3.5-turbo', api_key=os.environ["OPENAI_API_KEY"])

# Step-by-Step Reasoning을 위한 프롬프트 템플릿 정의
from langchain import PromptTemplate

step_prompt = PromptTemplate(
    input_variables=["question", "data"],  # 프롬프트에 입력될 변수들
    template='''
다음 문제를 주어진 데이터를 활용하여 단계별로 해결하세요. 각 단계의 결과를 간단히 기록하고, 최종 답변을 제공하세요.
문제: {question}
데이터: {data}

단계 1: 문제에서 요구하는 구매 항목과 수량을 나열하고, 데이터를 파악하여 개수 당 단가를 구하세요.
단계 2: 데이터에서 구한 단가를 활용하여 각 항목의 비용을 계산하고, 총액을 구하세요.
단계 3: 총액이 10,000원 이상인지 확인하고 해당 시 10% 할인을 적용하세요.
단계 4: 계산 결과를 검증하고, 최종 답변을 작성하세요. 단위는 '원' 이고 3자리 마다 콤마(,)를 찍어주세요.
'''
)

# 위에서 정의한 프롬프트와 LLM을 묶어 하나의 체인으로 생성
from langchain.chains import LLMChain
step_chain = LLMChain(llm=llm, prompt=step_prompt)

설명 요약

• LangChain 설치: 필요한 기능을 사용하기 위해 필수
• ChatOpenAI: OpenAI GPT 모델을 LangChain과 연동
• PromptTemplate: 문제 해결을 위한 구조화된 템플릿 설정
• LLMChain: 템플릿과 모델을 연결하여 실행 가능한 체인 구성

 

---

# LangChain + OpenAI LLM을 활용한 구매 금액 계산기
from typing import Any, Dict

# 구매 요청 데이터를 받아 문제 해결을 시도하는 함수
def solve_purchase_problem(request: PurchaseRequest) -> Dict[str, Any]:
    try:
        # 1. 질문 문자열 생성
        question = f"구매 항목: {request.items}, 총액은 얼마인가?"

        # 2. 프롬프트에 입력될 데이터 구성
        data = {
            "price": FRUIT_PRICE,  # 과일별 단가 및 개수당 환산 정보
            "discount": {
                "threshold": DISCOUNT_THRESHOLD,  # 할인 기준 금액
                "rate": DISCOUNT_RATE             # 할인율
            }
        }

        # 3. LangChain을 통해 프롬프트 실행 → GPT에게 단계별 reasoning 요청
        response = step_chain.invoke({
            'question': question,
            'data': str(data)  # 프롬프트에 문자열 형태로 전달
        })

        # 4. 응답 텍스트 추출 및 검증
        final_answer = response["text"]

        return {
            "status": "success",
            "question": question,
            "answer": final_answer
        }

    except Exception as e:
        # 예외 발생 시 에러 메시지 포함 반환
        return {
            "status": "error",
            "question": question,
            "answer": e
        }

# 테스트용 요청 객체 생성
request = PurchaseRequest(items={"사과": 6, "오렌지": 5, "바나나": 4})

# 문제 해결 함수 호출 및 결과 출력
result = solve_purchase_problem(request)
print(f"질문 : {result['question']}")
print(f"출력 : {result['answer']}")

코드 설명 요약

구성 요소	설명
question	LLM에게 던질 자연어 질문 구성 ("사과 6개, 오렌지 5개, ...")
data	프롬프트에 들어갈 추가 정보: 단가, 할인 기준 등
step_chain.invoke()	LangChain 체인을 실행하여 GPT 모델로부터 reasoning 기반 답변 받기
PurchaseRequest	Pydantic으로 정의된 입력 검증 클래스 (딕셔너리 형태의 과일 및 수량 입력)
예외 처리	모델 응답 실패, 포맷 오류 등의 예외를 처리해 사용자에게 오류 전달

 

---

멀티 프롬프트 라우팅 (Multi-Prompt Routing)

개념

하나의 LLM 시스템 내에서 사용자 입력의 목적이나 의도에 따라 서로 다른 프롬프트나 체인으로 분기 처리하는 방식입니다.

목적

• 다양한 유형의 요청(예: 계산, 추천, 조회 등)을 하나의 인터페이스에서 처리
• 각 요청에 최적화된 프롬프트 사용 → 정확도 향상
• 시스템 유연성과 유지 보수성 향상

---

예시: 과일 매장 애플리케이션

사용자 입력 예:

1. 사과 3개, 바나나 2개 샀을 때 비용은?
2. 바나나와 오렌지 중 뭐가 더 저렴해?
3. 요즘 인기 많은 과일 추천해줘
4. 오렌지 재고 있나요?

이런 요청을 다음과 같이 라우팅할 수 있음:

의도	분류 키워드	연결 프롬프트
비용 계산	"얼마", "비용", "총액", "가격" 등	비용 계산용 프롬프트 (step_prompt)
단가 비교	"더 비싸", "더 싸", "비교", "단가" 등	단가 비교 프롬프트
과일 추천	"추천", "인기", "요즘", "먹을까" 등	추천용 프롬프트
재고 확인	"재고", "있는지", "있나요" 등	재고 확인 프롬프트

 

---

구현 예시 (로직 스케치)

def route_prompt(user_input: str):
    if "얼마" in user_input or "비용" in user_input or "총액" in user_input:
        return "price"
    elif "비싸" in user_input or "싸" in user_input or "비교" in user_input:
        return "compare"
    elif "추천" in user_input or "인기" in user_input:
        return "recommend"
    elif "재고" in user_input or "있나요" in user_input:
        return "stock"
    else:
        return "default"

---

보충 설명: LangChain에서의 라우팅 체인

LangChain에서는 MultiPromptChain, RouterChain, Tool 등 다양한 구조를 사용할 수 있습니다.

• LLMRouterChain: 입력 분류를 위한 LLM 기반 라우팅 체인
• MultiPromptChain: 각 요청 유형에 대응되는 여러 개의 프롬프트 체인 묶음
• SimpleSelector: 키워드 기반 라우팅도 가능 (Lightweight)

---

정리

• 멀티 프롬프트 라우팅은 LLM 활용도를 높이고 다양한 작업을 하나의 시스템에서 처리하기 위한 전략
• 요청 의도를 정확히 파악하고 적절한 프롬프트를 적용해야 신뢰도 있는 응답 가능
• Rule 기반 라우팅 → LLM 기반 라우팅으로 발전 가능

 

---

from typing import Dict, Any, Optional  # 타입 힌트를 위한 표준 라이브러리

# 과일별 가격 정보 (킬로그램 단위 가격과 킬로그램당 개수)
FRUIT_PRICE = {
    "사과": {"price_per_kg": 7000, "units_per_kg": 4},   # 사과 1kg에 4개 → 1개 1750원
    "오렌지": {"price_per_kg": 8000, "units_per_kg": 5}, # 오렌지 1kg에 5개 → 1개 1600원
    "바나나": {"price_per_kg": 4500, "units_per_kg": 6}  # 바나나 1kg에 6개 → 1개 750원
}

# 현재 과일 재고 (개수 기준)
FRUIT_STOCK = {
    "사과": 10,
    "오렌지": 8,
    "바나나": 15
}

# 할인 조건 설정
DISCOUNT_THRESHOLD = 10000  # 1만 원 이상 구매 시
DISCOUNT_RATE = 0.1         # 10% 할인 적용

# 사용자 요청을 표현하기 위한 데이터 모델 정의
from pydantic import BaseModel, Field

class PurchaseRequest(BaseModel):  # 사용자 입력을 검증하고 구조화
    request_type : str = Field(..., description='요청 타입( 비용 계산, 단가 비교, 과일 추천, 재고 상태 확인 )')
    items: Optional[Dict[str, int]] = Field(None, description='과일 이름과 수량( 예: {"사과": 3, "오렌지": 4} )')
    budget: Optional[int] = Field(None, description='예산( 원 단위 )')
    compare_item : Optional[list[str]] = Field(None, description='비교할 과일 이름( 예: ["사과", "오렌지"] )')

전체 코드 설명:

1. FRUIT_PRICE:
   • 과일별로 킬로그램당 가격(price_per_kg)과 1킬로그램당 몇 개(units_per_kg)가 들어 있는지를 정의.
   • 이를 통해 개당 단가를 유도할 수 있음. 예: 사과 1개 = 7000 / 4 = 1750원.
2. FRUIT_STOCK:
   • 현재 각 과일의 재고 개수를 나타냄.
   • 사용자가 구매 가능 여부 확인 시 참조.
3. DISCOUNT_THRESHOLD / DISCOUNT_RATE:
   • 총 구매금액이 일정 금액을 넘을 경우 할인을 적용하기 위한 조건값.
   • 예: 10,000원 이상일 경우 10% 할인.
4. PurchaseRequest (Pydantic 모델):
   • 사용자 요청을 구조화된 형태로 받고 유효성을 검증하기 위해 사용.
   • request_type: 요청의 목적을 구분(예: 비용 계산, 재고 확인 등).
   • items: 과일과 수량 (예: {"사과": 3}).
   • budget: 예산 한도. 예: 추천할 때 예산에 맞는 과일만 선택.
   • compare_item: 단가 비교 요청 시 비교할 과일 목록.

req1 = PurchaseRequest(
    request_type = '비용계산',
    items = {'사과':3}
)
req1.items

 

---

 

# LLM 초기화
from langchain.chat_models import ChatOpenAI
import os
llm = ChatOpenAI(
    model='gpt-4o',
    api_key=os.environ["OPENAI_API_KEY"],
    temperature=0.1
    )

PROMPT_TEMPLATE = {
"비용계산" : """
  다음 문제를 주어진 데이터 와 단계별로 해결하세요, 각 단계의 결과를 간단히 기록하고, 최종답변을 제공하세요.
  문제 : {question}
  데이터 : {data}

  단계 1 : 문제에서 요구하는 구매 항목과 수량을 나열하고, 데이터보고 개수 당 단가를 계산하여 구하세요.
  단계 2 : 각 항목의 비용을 계산하고, 총액을 구하세요.
  단계 3 : 총액인 10,000원 이상인지 확인하고, 해당시 10% 할인을 적용하세요.
  단계 4 : 계산결과를 검증하고, 최종 답변을 작성하세요(단위는 원)
  """,
"단가비교": """
  다음 요청에 따라 과일 단가를 비교하세요. 데이터를 기반으로 겨롸를 간결하게 작성하세요.
  요청: {question}
  데이터: {data}

  단계 1: 요청된 과일의 단가를 데이터를 보고 구하세요.
  단계 2: 단가를 비교하고, 어떤 과일이 더 저렴한지 분석하세요.
  단계 3: 비교 결과를 명확히 작성하세요.( 단위: 원/개 )
  """,
"과일추천": """
  다음 예산 내에서 구매 가능한 과일을 추천하세요. 데이터를 기반으로 결과를 간결하게 작성하세요.
  예산: {question}
  데이터: {data}

  단계 1: 데이터를 보고 과일 단가와 재고 개수를 파악하세요.
  단게 2: 예산과 재고 내에서 구매 가능한 과일과 수량을 개수 단위로 계산하세요.
  단계 3: 할인가를 고려하여 가장 많은 수량을 구매할 수 있는 조합을 추천하고 총액을 계산하세요.
  단계 4: 추천 결과와 총액을 작성하세요. ( 단위: 원 )
  """,
"재고상태확인": """
  다음 요청에 따라 과일 재고를 확인하세요. 데이터를 기반으로 결과를 간결하게 작성하세요.
  요청: {question}
  데이터: {data}

  단계 1: 요청된 과일과 수량을 나열하고, 데이터에서 재고를 확인하세요.
  단계 2: 요청 수량이 재고를 초과하는지 확인하세요.
  단계 3: 구매 가능 여부와 재고 상태를 명확히 작성하세요.
  """
}

# 동적 프롬프트 체인 생성 함수
from langchain.chains import LLMChain
from langchain import PromptTemplate

def create_dynamic_chain(request_type: str) -> LLMChain:
    # 요청 타입이 미리 정의된 템플릿 딕셔너리(PROMPT_TEMPLATE)에 없으면 예외 발생
    if request_type not in PROMPT_TEMPLATE:
        raise ValueError(f"지원하지 않는 요청 타입입니다. {request_type}")
    
    # 해당 요청 타입에 맞는 프롬프트 템플릿을 가져와 PromptTemplate 객체 생성
    prompt = PromptTemplate(
        input_variables=["question", "data"],  # 프롬프트에서 사용할 변수
        template=PROMPT_TEMPLATE[request_type] # 템플릿 문자열
    )
    
    # PromptTemplate을 기반으로 LangChain의 LLMChain 생성 후 반환
    return LLMChain(llm=llm, prompt=prompt)

 

---

# 요청 라우팅 및 처리 함수
def process_user_request(request: PurchaseRequest) -> Dict[str, Any]:
    # 요청 유형에 따라 자연어 질문을 자동으로 생성
    if request.request_type == "비용계산":
        # 예: 구매 항목의 총액을 묻는 질문 생성
        question = f"구매 항목: {request.items} 총액은 얼마인가요?"
    elif request.request_type == "단가비교":
        # 예: 비교할 항목에 대한 단가 비교 질문 생성
        question = f"다음 과일의 단가를 비교: {request.compare_item}"
    elif request.request_type == "과일추천":
        # 예: 예산 내에서 추천 가능한 과일 질문 생성
        question = f"예산: {request.budget}원. 예산 내에서 구매 가능한 과일을 추천해주세요."
    elif request.request_type == "재고상태확인":
        # 예: 특정 과일의 재고 상태를 묻는 질문 생성
        question = f"구매 항목 : {request.items}"
    else:
        # 지원하지 않는 요청 유형일 경우 예외 처리
        raise ValueError(f"지원하지 않는 유형: {request.request_type}")

    # 데이터 준비 (가격, 재고, 할인 기준과 비율 등)
    data = {
        "prices": FRUIT_PRICE,
        "stock": FRUIT_STOCK,
        "discount": {
            "threshold": DISCOUNT_THRESHOLD,
            "rate": DISCOUNT_RATE
        }
    }

    # 요청 유형에 따라 적절한 동적 체인을 생성 (프롬프트 체인 혹은 기능 실행 체인)
    chain = create_dynamic_chain(request.request_type)

    # 생성된 체인을 통해 질문과 데이터를 전달하여 응답을 얻음
    result = chain.invoke({
        "question": question,
        "data": data  # 질문에 필요한 정보도 함께 전달
    })

    # 처리 결과 반환
    return result

코드 설명

이 함수는 PurchaseRequest라는 입력을 받아, 그 요청에 따라 적절한 질의 문장을 만들고, 그에 따라 응답을 생성하는 중앙 요청 처리 함수입니다.

동작 흐름 요약:

1. 요청 유형 파악 (request.request_type)
   • 비용계산: 항목들의 총액 계산 요청
   • 단가비교: 과일 간 단가 비교 요청
   • 과일추천: 예산 내 추천 요청
   • 재고상태확인: 특정 과일 재고 조회 요청
2. 사용자가 요청한 유형에 따라 다르게 처리함.
3. 질문(프롬프트) 생성
4. 각 유형별로 자연어 기반의 질의 문장을 생성함.
5. 데이터 구성
6. 과일 가격, 재고, 할인 정책 등의 정보를 딕셔너리로 준비.
7. 체인 구성 (create_dynamic_chain)
8. 요청 유형에 맞게 적절한 Chain(LangChain 스타일일 가능성 있음)을 선택 생성.
9. Chain 실행
10. chain.invoke()를 통해 질문과 데이터를 전달하고, 결과를 받아 반환.

 

---

purchase_data = PurchaseRequest(request_type="비용계산", items={"사과": 3, "오렌지": 5, "바나나": 6})
purchase_result = process_user_request(purchase_data)

compare_data = PurchaseRequest(request_type="단가비교", compare_item=["사과", "바나나"])
compare_result = process_user_request(compare_data)

recommand_data = PurchaseRequest(request_type="과일추천", budget=10000)
recommand_result = process_user_request(recommand_data)

status_data = PurchaseRequest(request_type="재고상태확인", items={"사과": 3, "오렌지": 5, "바나나": 6})
status_result = process_user_request(status_data)

print(f"질문: {purchase_result['question']}")
print(purchase_result["text"])
print('*'*100)

print(f"질문: {compare_result['question']}")
print(compare_result["text"])
print('*'*100)

print(f"질문: {recommand_result['question']}")
print(recommand_result["text"])
print('*'*100)

print(f"질문: {status_result['question']}")
print(status_result["text"])

질문: 구매 항목: {'사과': 3, '오렌지': 5, '바나나': 6} 총액은 얼마인가요?
### 단계 1: 구매 항목과 수량 나열 및 개수 당 단가 계산

- 구매 항목: 
  - 사과: 3개
  - 오렌지: 5개
  - 바나나: 6개

- 데이터에서 개수 당 단가 계산:
  - 사과: 7000원/4개 = 1750원/개
  - 오렌지: 8000원/5개 = 1600원/개
  - 바나나: 4500원/6개 = 750원/개

### 단계 2: 각 항목의 비용 계산 및 총액 구하기

- 사과 비용: 3개 × 1750원/개 = 5250원
- 오렌지 비용: 5개 × 1600원/개 = 8000원
- 바나나 비용: 6개 × 750원/개 = 4500원

- 총액: 5250원 + 8000원 + 4500원 = 17750원

### 단계 3: 총액이 10,000원 이상인지 확인 및 할인 적용

- 총액 17750원은 10,000원 이상이므로, 10% 할인 적용
- 할인 금액: 17750원 × 0.1 = 1775원
- 할인 후 총액: 17750원 - 1775원 = 15975원

### 단계 4: 계산 결과 검증 및 최종 답변 작성

- 계산 결과를 검증한 결과, 모든 단계가 정확하게 수행되었습니다.
- 최종 답변: 총액은 15,975원입니다.
****************************************************************************************************
질문: 다음 과일의 단가를 비교: ['사과', '바나나']
단계 1: 요청된 과일의 단가 계산

- 사과: 1kg에 4개, 가격은 7000원
  - 사과 1개의 가격 = 7000원 / 4개 = 1750원/개

- 바나나: 1kg에 6개, 가격은 4500원
  - 바나나 1개의 가격 = 4500원 / 6개 = 750원/개

단계 2: 단가 비교

- 사과 1개의 가격: 1750원
- 바나나 1개의 가격: 750원

단계 3: 비교 결과

- 바나나가 사과보다 더 저렴합니다. (사과: 1750원/개, 바나나: 750원/개)
****************************************************************************************************
질문: 예산: 10000원. 예산 내에서 구매 가능한 과일을 추천해주세요.
단계 1: 과일 단가와 재고 파악
- 사과: 1kg당 7000원, 1kg에 4개, 재고 10개
- 오렌지: 1kg당 8000원, 1kg에 5개, 재고 8개
- 바나나: 1kg당 4500원, 1kg에 6개, 재고 15개

단계 2: 예산과 재고 내에서 구매 가능한 과일과 수량 계산
- 사과: 10000원으로 1kg(4개) 구매 가능
- 오렌지: 10000원으로 1kg(5개) 구매 가능
- 바나나: 10000원으로 2kg(12개) 구매 가능

단계 3: 할인가를 고려하여 가장 많은 수량을 구매할 수 있는 조합 추천
- 바나나 2kg(12개) 구매 시 9000원
- 10000원 이상 구매 시 10% 할인 적용 가능
- 바나나 2kg 구매 시 9000원으로 할인 적용 불가

단계 4: 추천 결과와 총액 작성
- 추천 과일: 바나나
- 구매 수량: 12개
- 총액: 9000원
****************************************************************************************************
질문: 구매 항목 : {'사과': 3, '오렌지': 5, '바나나': 6}
**단계 1: 요청된 과일과 수량 나열 및 재고 확인**

- 사과: 요청 수량 3개, 재고 10개
- 오렌지: 요청 수량 5개, 재고 8개
- 바나나: 요청 수량 6개, 재고 15개

**단계 2: 요청 수량이 재고를 초과하는지 확인**

- 사과: 요청 수량 3개는 재고 10개보다 적음
- 오렌지: 요청 수량 5개는 재고 8개보다 적음
- 바나나: 요청 수량 6개는 재고 15개보다 적음

**단계 3: 구매 가능 여부와 재고 상태**

모든 요청된 과일의 수량이 재고 내에 있으므로 구매가 가능합니다. 각 과일의 재고는 충분합니다.

 

---

# json 응답 검증을 위한 pydantic 모델 정의
from pydantic import BaseModel, Field
from typing import Dict, Any, Optional

# '비용계산' 요청에 대한 응답 모델
class CalculationResponse(BaseModel):  # 결과 검증용 모델
    total_cost: Optional[float] = Field(None, description="총 비용")  # 총액
    currency: Optional[str] = Field(None, description="통화 단위")    # 예: KRW, USD
    details: Optional[Dict[str, Any]] = Field(None, description="상세 계산 내역")  # 품목별 비용 정보 등

# '단가비교' 요청에 대한 응답 모델
class CompareResponse(BaseModel):  # 결과 검증용 모델
    compare: Dict[str, float] = Field(..., description="과일별 단가 비교")  # 예: {'사과': 1000.0, '바나나': 800.0}
    cheapest: str = Field(..., description="가장 저렴한 과일")            # 예: '바나나'

# '과일추천' 요청에 대한 응답 모델
class RecoomendResponse(BaseModel):  # 결과 검증용 모델
    recommended_items: Dict[str, int] = Field(..., description="추천 과일과 수량")  # 예: {'사과': 2, '바나나': 1}
    total_cost: float = Field(..., description="총 비용")                         # 추천된 과일의 총액

# '재고상태확인' 요청에 대한 응답 모델
class StockResponse(BaseModel):  # 결과 검증용 모델
    stock_status: Dict[str, bool] = Field(..., description="과일별 구매 가능여부")  # 예: {'사과': True, '포도': False}
    available_stock: Dict[str, int] = Field(..., description="과일별 구매 가능 수량")  # 예: {'사과': 12, '포도': 0}

코드 설명

이 코드는 응답 데이터의 구조와 유효성 검증을 담당하는 Pydantic 모델 정의입니다. 각 요청 유형에 대해 응답이 JSON 형태로 오기 때문에, 그 내용을 자동으로 검증하고 구조화하기 위해 사용합니다.

각 클래스는 BaseModel을 상속받으며, 아래 목적을 가집니다.

1. CalculationResponse
   • 비용 계산 응답: 총 비용, 통화 단위, 항목별 상세 내역 등 포함.
   • Optional을 사용한 이유는 경우에 따라 필드가 누락될 수 있기 때문입니다.
2. CompareResponse
   • 단가 비교 응답: 과일별 가격과 가장 저렴한 과일 반환.
3. RecommendResponse
   • 과일 추천 응답: 추천된 과일별 수량과 총비용을 반환.
   • recommended_items는 예산 내 추천된 항목들을 의미합니다.
4. StockResponse
   • 재고 상태 응답: 구매 가능 여부와 수량을 반환합니다.

 

---

# 프롬프트 템플릿: 요청 유형별로 LLM에게 전달할 명령을 구조화된 형태로 정의
PROMPT_TEMPLATE_JSON = {
    "비용계산": '''
      다음 문제를 주어진 데이터와 단계별로 해결하세요. 각 단계의 결과를 간단히 기록하고, 최종 답변을 제공하세요.
      문제 : {question}
      데이터 : {data}

      단계 1 : 문제에서 요구하는 구매 항목과 수량을 나열하고, 데이터보고 개수 당 단가를 계산하여 구하세요.
      단계 2 : 각 항목의 비용을 계산하고, 총액을 구하세요.
      단계 3 : 총액이 10,000원 이상인지 확인하고, 해당 시 10% 할인을 적용하세요.
      단계 4 : 계산 결과를 검증하고, 최종 답변을 JSON 형식으로 작성하세요.
      ```json
      {{
        "total_cost": 총액,
        "currency": "원",
        "details": {{"items": 항목별 비용, "discount_applied": 할인 적용 여부, "discount_amount": 할인 금액}}
      }}
      ```
    ''',

    "단가비교": """
      다음 요청에 따라 과일 단가를 비교하세요. 데이터를 기반으로 결과를 간결하게 작성하세요.
      요청: {question}
      데이터: {data}

      단계 1: 요청된 과일의 단가를 데이터를 보고 구하세요.
      단계 2: 단가를 비교하고, 어떤 과일이 더 저렴한지 분석하세요.
      단계 3: 비교 결과를 JSON 형식으로 작성하세요. (단위: 원/개)
      ```json
      {{
        "compare": {{"과일": 단가}},
        "cheapest": "가장 저렴한 과일"
      }}
      ```
    """,

    "과일추천": """
      다음 예산 내에서 구매 가능한 과일 조합을 추천하세요. 데이터를 기반으로 결과를 간결하게 작성하세요.
      예산: {question}
      데이터: {data}

      단계 1: 데이터를 보고 과일의 단가와 재고 개수를 파악하세요.
      단계 2: 예산과 재고 내에서 구매 가능한 과일과 수량을 계산하세요.
      단계 3: 할인가를 고려하여 가장 많은 수량을 구매할 수 있는 조합을 추천하고 총액을 계산하세요.
      단계 4: 추천 결과와 총액을 JSON 형식으로 작성하세요. (단위: 원)
      ```json
      {{
        "recommended_items": {{"과일": 수량}},
        "total_cost": 총액,
        "currency": "원"
      }}
      ```
    """,

    "재고상태확인": """
      다음 요청에 따라 과일 재고를 확인하세요. 데이터를 기반으로 결과를 간결하게 작성하세요.
      요청: {question}
      데이터: {data}

      단계 1: 요청된 과일과 수량을 나열하고, 데이터에서 재고를 확인하세요.
      단계 2: 요청 수량이 재고를 초과하는지 확인하세요.
      단계 3: 구매 가능 여부와 재고 상태를 JSON 형식으로 작성하세요.
      ```json
      {{
        "stock_status": {{"과일": 구매 가능 여부}},
        "available_stock": {{"과일": 구매 가능 수량}}
      }}
      ```
    """
}

# LLM 초기화: OpenAI GPT-3.5 모델을 LangChain에서 사용하도록 설정
from langchain.chat_models import ChatOpenAI
import os

llm = ChatOpenAI(
    model='gpt-3.5-turbo',             # 사용할 모델명
    api_key=os.environ["OPENAI_API_KEY"],  # 환경변수에서 API 키 불러오기
    temperature=0.1                    # 응답의 일관성을 위해 낮은 온도 설정
)

# 동적 프롬프트 체인 생성 함수: 요청 유형에 맞는 템플릿을 기반으로 LLMChain 객체를 생성
from langchain.chains import LLMChain
from langchain import PromptTemplate

def create_dynamic_chain_json(request_type: str) -> LLMChain:
    # 정의된 요청 유형이 아니면 예외 발생
    if request_type not in PROMPT_TEMPLATE_JSON:
        raise ValueError(f"지원하지 않는 요청 타입입니다. {request_type}")

    # LangChain용 프롬프트 템플릿 생성
    prompt = PromptTemplate(
        input_variables=["question", "data"],                   # 템플릿에서 사용할 변수
        template=PROMPT_TEMPLATE_JSON[request_type]             # 해당 요청 유형의 템플릿 선택
    )

    # LLM과 프롬프트를 묶은 체인 반환
    return LLMChain(llm=llm, prompt=prompt)

전체 설명

이 코드는 사용자의 요청 유형에 따라 적절한 프롬프트를 생성하고, 해당 프롬프트를 기반으로 OpenAI LLM을 호출할 수 있는 LangChain의 LLMChain 객체를 생성합니다.

핵심 구성 요소

• PROMPT_TEMPLATE_JSON: LLM에게 보낼 질문 형식을 요청 유형별로 구조화한 딕셔너리입니다.
• ChatOpenAI: OpenAI의 GPT-3.5-turbo 모델을 사용할 수 있도록 LangChain에 등록합니다.
• PromptTemplate: LangChain의 포맷팅 템플릿 객체로, question과 data를 주입할 수 있습니다.
• create_dynamic_chain_json(): 요청 유형에 맞는 템플릿을 사용해 LLMChain 객체를 만들어 반환합니다.

 

import json
from typing import Dict, Any

# JSON 응답 파싱 및 유효성 검증 함수
def parse_and_validate_response(request_type: str, response_text: str) -> Dict[str, Any]:
    # LLM 응답에서 ```json ... ``` 블록을 추출
    json_start = response_text.find("```json")
    json_end = response_text.find("```", json_start + 1)
    if json_start == -1 or json_end == -1:
        raise ValueError("JSON 형식이 응답에 포함되지 않았습니다.")
    
    # JSON 문자열만 추출
    json_str = response_text[json_start + len("```json"):json_end]
    result_json = json.loads(json_str)  # 문자열을 dict로 파싱

    # 요청 유형에 따라 데이터 검증: 각 유형에 맞는 Pydantic 모델 사용
    if request_type == "비용계산":
        validated = CalculationResponse(**result_json).model_dump()
    elif request_type == "단가비교":
        validated = CompareResponse(**result_json).model_dump()
    elif request_type == "과일추천":
        validated = RecommendResponse(**result_json).model_dump()
    elif request_type == "재고상태확인":
        validated = StockResponse(**result_json).model_dump()
    else:
        raise ValueError(f"지원하지 않는 유형: {request_type}")
    
    # 검증된 데이터와 원시 응답 모두 반환
    return {
        'status': "success",
        "parsed_response": validated,   # 검증된 응답
        "raw_response": result_json     # 파싱된 원시 응답
    }

 

# 사용자 요청 처리 및 JSON 응답 검증 라우팅 함수
def process_user_request_json(request: PurchaseRequest) -> Dict[str, Any]:
    # 요청 유형에 따라 질문 프롬프트 생성
    if request.request_type == "비용계산":
        question = f"구매 항목: {request.items} 총액은 얼마인가요?"
    elif request.request_type == "단가비교":
        question = f"다음 과일의 단가를 비교: {request.compare_item}"
    elif request.request_type == "과일추천":
        question = f"예산: {request.budget}원. 예산 내에서 구매 가능한 과일 조합을 추천해주세요."
    elif request.request_type == "재고상태확인":
        question = f"구매 항목 : {request.items}"
    else:
        raise ValueError(f"지원하지 않는 유형: {request.request_type}")

    # 모델에 제공할 데이터 구성
    data = {
        "prices": FRUIT_PRICE,  # 과일 가격 정보
        "stock": FRUIT_STOCK,   # 과일 재고 정보
        "discount": {
            "threshold": DISCOUNT_THRESHOLD,  # 할인 기준 금액
            "rate": DISCOUNT_RATE             # 할인율
        }
    }

    # 요청 유형에 맞는 프롬프트 체인 생성 (JSON 응답용)
    chain = create_dynamic_chain_json(request.request_type)

    # LLM 체인 실행: 프롬프트(question)와 데이터(data)를 입력
    result = chain.invoke({"question": question, "data": data})

    # LLM 응답(JSON)을 파싱하고 유효성 검증
    parser_data = parse_and_validate_response(request.request_type, result["text"])

    # 원본 LLM 응답(result)은 유지하고, 필요시 parser_data를 합쳐서 반환 가능
    return result

 

purchase_data = PurchaseRequest(request_type="비용계산", items={"사과": 3, "오렌지": 5, "바나나":10})
purchase_result = process_user_request_json(purchase_data)

print(purchase_result["question"])
print(purchase_result["text"])

구매 항목: {'사과': 3, '오렌지': 5, '바나나': 10} 총액은 얼마인가요?
**단계 1:**
- 구매 항목: {'사과': 3, '오렌지': 5, '바나나': 10}
- 사과: 개당 가격 7000원, 1kg 당 4개
- 오렌지: 개당 가격 8000원, 1kg 당 5개
- 바나나: 개당 가격 4500원, 1kg 당 6개

**단계 2:**
- 사과: (3 / 4) * 7000 = 5250원
- 오렌지: (5 / 5) * 8000 = 8000원
- 바나나: (10 / 6) * 4500 = 7500원
- 총액: 5250 + 8000 + 7500 = 20750원

**단계 3:**
- 총액이 10,000원 이상이므로 10% 할인 적용
- 할인 금액: 20750 * 0.1 = 2075원
- 최종 총액: 20750 - 2075 = 18675원

**단계 4:**
```json
{
  "total_cost": 18675,
  "currency": "원",
  "details": {
    "items": {"사과": 5250, "오렌지": 8000, "바나나": 7500},
    "discount_applied": true,
    "discount_amount": 2075
  }
}