LLM 애플리케이션 개발 가이드: 프로토타입에서 프로덕션까지

┌─────────────────────────────────────────┐
│           사용자 인터페이스              │
│  (Web, Mobile, API, Slack, CLI)         │
├─────────────────────────────────────────┤
│           애플리케이션 로직              │
│  (오케스트레이션, 비즈니스 규칙)         │
├─────────────────────────────────────────┤
│        LLM 오케스트레이션 레이어         │
│  (LangChain, LlamaIndex, 원시 SDK)      │
├─────────────────────────────────────────┤
│           LLM 제공자                    │
│  (OpenAI, Anthropic, Google, 로컬)      │
├─────────────────────────────────────────┤
│           지원 서비스                   │
│  (벡터 DB, 캐시, 검색, 툴)              │
└─────────────────────────────────────────┘

[시스템 지침]
당신은 Acme Corp의 친절한 고객 지원 에이전트입니다.
사용자가 쓰는 언어로 응답하세요.
항상 예의 바르고 간결하게. 경쟁사는 절대 언급하지 마세요.

[컨텍스트 / 검색된 문서]
주문 #12345, 2026-03-10 접수. 상태: 발송됨.
운송장 번호: 1Z999AA10123456784

[예시 (few-shot)]
사용자: 제 주문은 어디 있나요?
어시스턴트: 주문 #99999가 3월 5일에 발송되어 배송 중입니다.
예상 배달일: 3월 12일.

[사용자 메시지]
지난주에 주문했는데 아직 받지 못했어요.

SYSTEM_PROMPT = """당신은 핀테크 회사의 시니어 Python 코드 리뷰어입니다.

책임:
- 정확성, 보안 취약점, 성능 문제를 검토
- 코드 예시를 포함한 구체적인 개선 사항 제안
- GDPR/CCPA를 위반하는 PII 처리 식별

출력 형식:
- 한 문장의 전반적인 평가로 시작
- 심각도와 함께 문제 목록: [CRITICAL], [WARNING], [SUGGESTION]
- 변경이 필요한 경우 수정된 코드 블록으로 마무리

새로운 기능을 생성하지 않습니다. 주어진 것만 검토합니다."""

FEW_SHOT_EXAMPLES = """
아래 회의록에서 액션 아이템을 추출하세요.
JSON 배열로 출력하세요.

회의록: John이 금요일까지 배포 가이드를 업데이트할 예정입니다.
Sarah는 이사회 회의 전에 Q1 예산을 검토해야 합니다.
액션 아이템: [
  {"owner": "John", "task": "배포 가이드 업데이트", "due": "금요일"},
  {"owner": "Sarah", "task": "Q1 예산 검토", "due": "이사회 회의 전"}
]

회의록: API 팀이 이번 스프린트에 속도 제한을 추가하기로 합의했습니다.
문서 업데이트는 담당자가 지정되지 않았습니다.
액션 아이템: [
  {"owner": "API team", "task": "속도 제한 추가", "due": "이번 스프린트"},
  {"owner": null, "task": "문서 업데이트", "due": null}
]

회의록: {meeting_text}
액션 아이템:"""

COT_PROMPT = """다음 문제를 단계별로 풀어보세요.
각 단계에서 추론 과정을 보여준 다음 최종 답변을 제시하세요.

문제: 고객의 크레딧이 500달러입니다. 320달러짜리 주문을 하고,
80달러짜리 상품 하나를 반품했습니다. 남은 크레딧은 얼마인가요?

단계별로 생각해봅시다:"""

from openai import OpenAI
import json

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o",
    response_format={"type": "json_object"},
    messages=[
        {"role": "system", "content": "엔티티를 추출하세요. 유효한 JSON만 출력하세요."},
        {"role": "user", "content": "Apple이 2024년 9월 9일 쿠퍼티노에서 iPhone 16을 발표했습니다."}
    ]
)

data = json.loads(response.choices[0].message.content)
# {"company": "Apple", "product": "iPhone 16", "location": "쿠퍼티노", "date": "2024-09-09"}

from pydantic import BaseModel
from openai import OpenAI

class NewsEvent(BaseModel):
    company: str
    product: str
    location: str
    date: str

client = OpenAI()
response = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[
        {"role": "system", "content": "이벤트 세부 사항을 추출하세요."},
        {"role": "user", "content": "Apple이 2024년 9월 9일 쿠퍼티노에서 iPhone 16을 발표했습니다."}
    ],
    response_format=NewsEvent,
)
event = response.choices[0].message.parsed
print(event.company)  # Apple

from openai import OpenAI

client = OpenAI(api_key="sk-...")  # 또는 OPENAI_API_KEY 환경 변수 설정

# 기본 채팅 완성
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "당신은 도움이 되는 어시스턴트입니다."},
        {"role": "user", "content": "트랜스포머 아키텍처를 3문장으로 요약해줘."}
    ],
    temperature=0.7,
    max_tokens=200,
)

print(response.choices[0].message.content)
print(f"사용된 토큰: {response.usage.total_tokens}")

import anthropic

client = anthropic.Anthropic(api_key="sk-ant-...")

message = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    system="당신은 도움이 되는 어시스턴트입니다.",
    messages=[
        {"role": "user", "content": "트랜스포머의 어텐션 메커니즘을 설명해줘."}
    ]
)

print(message.content[0].text)
print(f"입력 토큰: {message.usage.input_tokens}")
print(f"출력 토큰: {message.usage.output_tokens}")

from litellm import completion

# OpenAI
response = completion(
    model="gpt-4o",
    messages=[{"role": "user", "content": "안녕하세요"}]
)

# Anthropic (동일한 인터페이스)
response = completion(
    model="anthropic/claude-opus-4-5",
    messages=[{"role": "user", "content": "안녕하세요"}]
)

# 로컬 Ollama 모델 (동일한 인터페이스)
response = completion(
    model="ollama/llama3",
    messages=[{"role": "user", "content": "안녕하세요"}]
)

print(response.choices[0].message.content)

class ConversationManager:
    def __init__(self, system_prompt: str, max_history: int = 20):
        self.system_prompt = system_prompt
        self.max_history = max_history
        self.history: list[dict] = []
        self.client = OpenAI()

    def chat(self, user_message: str) -> str:
        self.history.append({"role": "user", "content": user_message})

        # 컨텍스트 윈도우 초과 방지를 위해 히스토리 트리밍
        if len(self.history) > self.max_history:
            self.history = self.history[-self.max_history:]

        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": self.system_prompt},
                *self.history
            ]
        )

        assistant_message = response.choices[0].message.content
        self.history.append({"role": "assistant", "content": assistant_message})
        return assistant_message

사용자 질의
    │
    ▼
[질의 임베딩] ──► [벡터 검색] ──► 상위 K개 관련 청크
                      │
                      ▼
            [증강된 프롬프트 구성]
            시스템: 도움이 되는 어시스턴트입니다.
            컨텍스트: {검색된 청크}
            사용자: {원래 질의}
                      │
                      ▼
               [LLM이 답변 생성]

from langchain.document_loaders import PyPDFLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# 1. 문서 로드
loader = DirectoryLoader("./docs", glob="**/*.pdf", loader_cls=PyPDFLoader)
documents = loader.load()

# 2. 청크로 분할
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", " ", ""]
)
chunks = splitter.split_documents(documents)

# 3. 임베딩 및 저장
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)
print(f"{len(documents)}개 문서에서 {len(chunks)}개 청크 인덱싱 완료")

from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

# 기존 벡터스토어 로드
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings(model="text-embedding-3-small")
)

# 검색기 생성
retriever = vectorstore.as_retriever(
    search_type="mmr",          # 다양성을 위한 Maximal Marginal Relevance
    search_kwargs={"k": 5, "fetch_k": 20}
)

# 커스텀 프롬프트
QA_PROMPT = PromptTemplate(
    template="""다음 컨텍스트를 사용하여 질문에 답하세요.
컨텍스트에 답이 없으면 "해당 정보가 없습니다."라고 말하세요.
정보를 만들어내지 마세요.

컨텍스트:
{context}

질문: {question}

답변:""",
    input_variables=["context", "question"]
)

# 체인
llm = ChatOpenAI(model="gpt-4o", temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    chain_type_kwargs={"prompt": QA_PROMPT},
    return_source_documents=True
)

result = qa_chain.invoke({"query": "환불 정책은 무엇인가요?"})
print(result["result"])
for doc in result["source_documents"]:
    print(f"출처: {doc.metadata['source']}, 페이지 {doc.metadata.get('page', 'N/A')}")

from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever

# 밀집 검색기 (시맨틱)
dense_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 희소 검색기 (BM25 키워드)
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 5

# 동일 가중치로 앙상블
ensemble_retriever = EnsembleRetriever(
    retrievers=[dense_retriever, bm25_retriever],
    weights=[0.6, 0.4]
)

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank(query: str, docs: list, top_k: int = 3) -> list:
    pairs = [(query, doc.page_content) for doc in docs]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(scores, docs), key=lambda x: x[0], reverse=True)
    return [doc for _, doc in ranked[:top_k]]

import json
import requests
from openai import OpenAI

client = OpenAI()

# 툴 정의
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "도시의 현재 날씨를 가져옵니다",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "도시명, 예: '서울'"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "온도 단위"
                    }
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "최신 정보를 위해 웹을 검색합니다",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "검색 쿼리"}
                },
                "required": ["query"]
            }
        }
    }
]

def get_weather(city: str, unit: str = "celsius") -> dict:
    # 실제 구현에서는 날씨 API를 호출
    return {"city": city, "temp": 18, "unit": unit, "condition": "맑음"}

def search_web(query: str) -> str:
    # 실제 구현에서는 검색 API를 호출
    return f"검색 결과: {query}"

TOOL_MAP = {
    "get_weather": get_weather,
    "search_web": search_web,
}

def run_agent(user_message: str) -> str:
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto"
        )

        message = response.choices[0].message

        # 툴 호출 없음 → 최종 답변
        if not message.tool_calls:
            return message.content

        # 히스토리에 어시스턴트 응답 추가
        messages.append(message)

        # 각 툴 호출 실행
        for tool_call in message.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)

            result = TOOL_MAP[func_name](**func_args)

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result)
            })

answer = run_agent("도쿄 날씨가 어떤가요? 우산을 가져가야 할까요?")
print(answer)

# 모델이 여러 툴을 동시에 호출할 수 있습니다
# 위의 루프가 이미 이를 처리합니다 — message.tool_calls는 리스트입니다
# 날씨와 검색을 단일 모델 턴에서 모두 호출 가능

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "도쿄와 파리의 날씨를 비교하고, 여행 팁도 검색해줘."}
    ],
    tools=tools,
    parallel_tool_calls=True  # GPT-4o에서 기본값 True
)

# 응답에 3개의 툴 호출이 동시에 있을 수 있습니다: weather(도쿄), weather(파리), search(여행 팁)

from openai import OpenAI

client = OpenAI()

# 동기 스트리밍
with client.chat.completions.stream(
    model="gpt-4o",
    messages=[{"role": "user", "content": "로봇에 관한 짧은 이야기를 써줘."}]
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI()

@app.post("/chat")
async def chat(body: dict):
    async def generate():
        async with client.chat.completions.stream(
            model="gpt-4o",
            messages=body["messages"]
        ) as stream:
            async for text in stream.text_stream:
                yield f"data: {text}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def classify_one(text: str, semaphore: asyncio.Semaphore) -> str:
    async with semaphore:
        response = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "POSITIVE, NEGATIVE, 또는 NEUTRAL로 분류하세요."},
                {"role": "user", "content": text}
            ],
            max_tokens=10
        )
        return response.choices[0].message.content.strip()

async def classify_batch(texts: list[str], max_concurrent: int = 20) -> list[str]:
    semaphore = asyncio.Semaphore(max_concurrent)
    tasks = [classify_one(text, semaphore) for text in texts]
    return await asyncio.gather(*tasks)

# 사용 예시
texts = ["이 제품 정말 좋아요!", "끔찍한 경험이었습니다.", "그냥 그랬어요."] * 100
results = asyncio.run(classify_batch(texts))

assert add(2, 3) == 5  # 항상 통과 또는 실패

from openai import OpenAI

client = OpenAI()

JUDGE_PROMPT = """AI 어시스턴트의 응답을 평가하고 있습니다.
다음 기준으로 응답을 평가하세요 (각 1-5점):
- 정확성: 정보가 맞나요?
- 유용성: 질문을 완전히 다루고 있나요?
- 간결성: 적절히 간결한가요?

질문: {question}
응답: {response}
참조 답변: {reference}

JSON으로 출력: {{"accuracy": X, "helpfulness": X, "conciseness": X, "reasoning": "..."}}"""

def evaluate(question: str, response: str, reference: str) -> dict:
    import json
    result = client.chat.completions.create(
        model="gpt-4o",
        response_format={"type": "json_object"},
        messages=[{
            "role": "user",
            "content": JUDGE_PROMPT.format(
                question=question,
                response=response,
                reference=reference
            )
        }]
    )
    return json.loads(result.choices[0].message.content)

from deepeval import evaluate
from deepeval.metrics import (
    AnswerRelevancyMetric,
    FaithfulnessMetric,
    ContextualRecallMetric,
)
from deepeval.test_case import LLMTestCase

test_case = LLMTestCase(
    input="프랑스의 수도는 어디인가요?",
    actual_output="프랑스의 수도는 파리입니다.",
    expected_output="파리",
    retrieval_context=["프랑스는 서유럽의 나라입니다. 수도는 파리입니다."]
)

metrics = [
    AnswerRelevancyMetric(threshold=0.8),
    FaithfulnessMetric(threshold=0.9),
    ContextualRecallMetric(threshold=0.8),
]

evaluate([test_case], metrics)

# promptfooconfig.yaml
prompts:
  - '다음 텍스트를 2문장으로 요약하세요: {{text}}'

providers:
  - openai:gpt-4o
  - openai:gpt-4o-mini

tests:
  - vars:
      text: '에펠탑은 1889년 만국박람회를 위해 건설되었습니다...'
    assert:
      - type: llm-rubric
        value: '요약에는 1889년과 만국박람회가 언급되어야 합니다'
      - type: javascript
        value: "output.split('.').length <= 3" # 최대 3문장

import tiktoken

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

def estimate_cost(
    input_tokens: int,
    output_tokens: int,
    model: str = "gpt-4o"
) -> float:
    # 100만 토큰당 가격 (2026년 3월 기준 대략적인 수치)
    PRICING = {
        "gpt-4o": {"input": 2.50, "output": 10.00},
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "claude-opus-4-5": {"input": 15.00, "output": 75.00},
        "claude-haiku-3-5": {"input": 0.80, "output": 4.00},
    }
    p = PRICING.get(model, {"input": 5.0, "output": 15.0})
    return (input_tokens * p["input"] + output_tokens * p["output"]) / 1_000_000

# Anthropic 프롬프트 캐싱
import anthropic

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": very_long_system_prompt,  # 예: 50K 토큰 정책 문서
            "cache_control": {"type": "ephemeral"}  # 이 접두사 캐싱
        }
    ],
    messages=[{"role": "user", "content": user_question}]
)
# 첫 번째 호출: 전체 가격. 이후 호출: 캐시된 토큰 약 90% 할인.

def route_to_model(task: str, complexity: str) -> str:
    """작업 복잡도에 따라 적절한 모델로 라우팅."""
    if complexity == "simple":
        return "gpt-4o-mini"          # 단순 분류, 추출
    elif complexity == "medium":
        return "gpt-4o"               # 요약, Q&A
    else:
        return "claude-opus-4-5"      # 복잡한 추론, 코드 리뷰

# 예시: 라우팅 전 복잡도 분류
def smart_complete(messages: list, task_description: str) -> str:
    from openai import OpenAI
    client = OpenAI()

    # 저렴한 분류 단계
    complexity = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"이 작업의 복잡도를 'simple', 'medium', 'complex' 중 하나로 평가해줘: {task_description}"
        }],
        max_tokens=5
    ).choices[0].message.content.strip().lower()

    model = route_to_model(task_description, complexity)

    return client.chat.completions.create(
        model=model,
        messages=messages
    ).choices[0].message.content

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from openai import AsyncOpenAI
import asyncio

app = FastAPI(title="LLM API")
client = AsyncOpenAI()

class ChatRequest(BaseModel):
    messages: list[dict]
    model: str = "gpt-4o"
    temperature: float = 0.7
    max_tokens: int = 1000

class ChatResponse(BaseModel):
    content: str
    usage: dict

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    try:
        response = await client.chat.completions.create(
            model=request.model,
            messages=request.messages,
            temperature=request.temperature,
            max_tokens=request.max_tokens,
        )
        return ChatResponse(
            content=response.choices[0].message.content,
            usage={
                "input_tokens": response.usage.prompt_tokens,
                "output_tokens": response.usage.completion_tokens
            }
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

import asyncio
import random
from functools import wraps

def with_retry(max_attempts: int = 3, base_delay: float = 1.0):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    # 지터가 있는 지수 백오프
                    delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                    await asyncio.sleep(delay)
        return wrapper
    return decorator

@with_retry(max_attempts=3)
async def robust_completion(messages: list) -> str:
    client = AsyncOpenAI()
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=messages
    )
    return response.choices[0].message.content

import hashlib
import json
import redis.asyncio as redis

redis_client = redis.from_url("redis://localhost:6379")
CACHE_TTL = 3600  # 1시간

def cache_key(messages: list, model: str) -> str:
    payload = json.dumps({"messages": messages, "model": model}, sort_keys=True)
    return f"llm:{hashlib.md5(payload.encode()).hexdigest()}"

async def cached_completion(messages: list, model: str = "gpt-4o") -> str:
    key = cache_key(messages, model)

    # 캐시 확인
    cached = await redis_client.get(key)
    if cached:
        return cached.decode()

    # 생성
    from openai import AsyncOpenAI
    client = AsyncOpenAI()
    response = await client.chat.completions.create(model=model, messages=messages)
    result = response.choices[0].message.content

    # TTL과 함께 저장
    await redis_client.setex(key, CACHE_TTL, result)
    return result

import os
from langchain_openai import ChatOpenAI
from langchain.callbacks.tracers import LangChainTracer

os.environ["LANGCHAIN_API_KEY"] = "ls__..."
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "my-llm-app"

# 모든 LangChain 호출이 자동으로 추적됨
llm = ChatOpenAI(model="gpt-4o")
response = llm.invoke("RAG란 무엇인가요?")
# LangSmith UI에서 전체 추적(프롬프트, 응답, 레이턴시, 토큰) 확인 가능

import time
import logging
from dataclasses import dataclass, field, asdict

logger = logging.getLogger(__name__)

@dataclass
class LLMCallLog:
    model: str
    input_tokens: int
    output_tokens: int
    latency_ms: float
    success: bool
    error: str = ""
    metadata: dict = field(default_factory=dict)

async def traced_completion(messages: list, model: str = "gpt-4o", **metadata) -> str:
    from openai import AsyncOpenAI
    client = AsyncOpenAI()

    start = time.perf_counter()
    success = True
    error = ""
    input_tokens = output_tokens = 0

    try:
        response = await client.chat.completions.create(model=model, messages=messages)
        result = response.choices[0].message.content
        input_tokens = response.usage.prompt_tokens
        output_tokens = response.usage.completion_tokens
        return result
    except Exception as e:
        success = False
        error = str(e)
        raise
    finally:
        log = LLMCallLog(
            model=model,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            latency_ms=(time.perf_counter() - start) * 1000,
            success=success,
            error=error,
            metadata=metadata
        )
        logger.info("llm_call", extra=asdict(log))

from guardrails import Guard
from guardrails.hub import ToxicLanguage, DetectPII

guard = Guard().use_many(
    ToxicLanguage(threshold=0.5, on_fail="exception"),
    DetectPII(pii_entities=["EMAIL_ADDRESS", "PHONE_NUMBER"], on_fail="fix"),
)

def safe_completion(user_input: str) -> str:
    from openai import OpenAI
    client = OpenAI()

    # 입력 검증
    guard.validate(user_input)

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": user_input}]
    )
    output = response.choices[0].message.content

    # 출력 검증 및 수정
    validated = guard.validate(output)
    return validated.validated_output