LLM 애플리케이션 개발 실전 가이드
LLM API를 처음 사용하면 "Hello World"는 금방 만들 수 있지만, 실제 프로덕션 서비스를 개발할 때는 수많은 함정이 있습니다. 스트리밍 구현, 비용 폭탄 방지, 속도 제한 처리, 할루시네이션 최소화... 이 가이드에서는 OpenAI, Anthropic Claude, Google Gemini API를 활용한 실전 LLM 애플리케이션 개발의 모든 것을 다룹니다.
1. LLM API 생태계 개요
주요 API 제공사 비교
| 제공사 | 주요 모델 | 강점 | 약점 |
| ----------- | ---------------------- | --------------------- | ---------------- |
| OpenAI | GPT-4o, o1, o3 | 생태계, 함수 호출 | 비용 |
| Anthropic | Claude 3.5, Claude 3.7 | 긴 컨텍스트, 안전성 | 상대적 높은 비용 |
| Google | Gemini 1.5 Pro/Flash | 멀티모달, 1M 컨텍스트 | 한국어 |
| Mistral AI | Mistral Large/NeMo | 비용 효율 | 생태계 |
| Cohere | Command R+ | 기업용, RAG | 제한된 기능 |
| Together AI | 오픈소스 모델 호스팅 | 오픈소스 모델 접근 | 신뢰성 |
API 비용 비교 (2025년 기준, 1M 토큰당)
API 비용 비교 (입력/출력 토큰 USD)
api_costs = {
"gpt-4o": {"input": 2.50, "output": 10.00},
"gpt-4o-mini": {"input": 0.15, "output": 0.60},
"claude-3-5-sonnet": {"input": 3.00, "output": 15.00},
"claude-3-haiku": {"input": 0.25, "output": 1.25},
"gemini-1.5-pro": {"input": 1.25, "output": 5.00},
"gemini-1.5-flash": {"input": 0.075, "output": 0.30},
"mistral-large": {"input": 2.00, "output": 6.00},
"mistral-small": {"input": 0.20, "output": 0.60},
}
def estimate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
"""API 비용 예상 계산"""
if model not in api_costs:
return 0.0
costs = api_costs[model]
input_cost = (input_tokens / 1_000_000) * costs["input"]
output_cost = (output_tokens / 1_000_000) * costs["output"]
return input_cost + output_cost
예시: 하루 10,000번 요청, 평균 200 입력 / 500 출력 토큰
daily_cost_gpt4o = estimate_cost("gpt-4o", 200 * 10000, 500 * 10000)
daily_cost_mini = estimate_cost("gpt-4o-mini", 200 * 10000, 500 * 10000)
print(f"GPT-4o 일일 비용: ${daily_cost_gpt4o:.2f}")
print(f"GPT-4o-mini 일일 비용: ${daily_cost_mini:.2f}")
2. OpenAI API 완전 가이드
기본 설정
pip install openai
from openai import OpenAI
클라이언트 초기화
client = OpenAI(
api_key=os.environ.get("OPENAI_API_KEY"),
기본값이지만 커스텀 가능
max_retries=3,
timeout=60.0,
)
기본 Chat Completions
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": "당신은 친절하고 정확한 한국어 AI 어시스턴트입니다."
},
{
"role": "user",
"content": "파이썬의 장단점을 알려주세요."
}
],
temperature=0.7, # 0: 결정론적, 1: 창의적
max_tokens=1024, # 최대 출력 토큰
top_p=0.9, # nucleus sampling
frequency_penalty=0.0, # 반복 단어 패널티
presence_penalty=0.0, # 새 토픽 도입 패널티
)
print(response.choices[0].message.content)
print(f"사용 토큰 - 입력: {response.usage.prompt_tokens}, 출력: {response.usage.completion_tokens}")
스트리밍 응답
def stream_chat(messages: list, model: str = "gpt-4o") -> str:
"""스트리밍 응답 처리"""
full_response = ""
with client.chat.completions.create(
model=model,
messages=messages,
stream=True,
) as stream:
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
print(delta.content, end="", flush=True)
full_response += delta.content
print() # 줄바꿈
return full_response
사용 예시
messages = [{"role": "user", "content": "머신러닝의 종류를 설명해주세요."}]
response = stream_chat(messages)
Function Calling (도구 활용)
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": "검색 쿼리"
},
"num_results": {
"type": "integer",
"description": "반환할 결과 수",
"default": 5
}
},
"required": ["query"]
}
}
}
]
def get_weather(city: str, unit: str = "celsius") -> dict:
"""날씨 API 호출 (실제 구현 필요)"""
실제로는 날씨 API 호출
return {
"city": city,
"temperature": 22,
"unit": unit,
"condition": "맑음",
"humidity": 45
}
def search_web(query: str, num_results: int = 5) -> list:
"""웹 검색 (실제 구현 필요)"""
return [{"title": f"검색 결과 {i}", "url": f"https://example.com/{i}"} for i in range(num_results)]
def run_conversation(user_message: str) -> str:
messages = [
{"role": "system", "content": "당신은 날씨와 정보 검색을 도와주는 AI 어시스턴트입니다."},
{"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
messages.append(message)
도구 호출이 없으면 응답 반환
if not message.tool_calls:
return message.content
도구 호출 처리
for tool_call in message.tool_calls:
function_name = tool_call.function.name
function_args = json.loads(tool_call.function.arguments)
if function_name == "get_weather":
result = get_weather(**function_args)
elif function_name == "search_web":
result = search_web(**function_args)
else:
result = {"error": "알 수 없는 함수"}
도구 결과 추가
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result, ensure_ascii=False)
})
사용 예시
print(run_conversation("서울과 부산의 현재 날씨를 비교해주세요."))
Structured Outputs (구조화된 출력)
from pydantic import BaseModel
from typing import List, Optional
class ProductReview(BaseModel):
product_name: str
overall_rating: int # 1-5
pros: List[str]
cons: List[str]
summary: str
recommendation: bool
def analyze_review(review_text: str) -> ProductReview:
response = client.beta.chat.completions.parse(
model="gpt-4o",
messages=[
{
"role": "system",
"content": "제품 리뷰를 분석하여 구조화된 정보로 추출하세요."
},
{
"role": "user",
"content": review_text
}
],
response_format=ProductReview,
)
return response.choices[0].message.parsed
사용 예시
review = """
삼성 갤럭시 S24를 2달 동안 사용해봤습니다.
카메라 품질이 정말 훌륭하고 배터리도 하루 종일 쓰기에 충분합니다.
AI 기능들도 실용적으로 잘 활용하고 있어요.
다만 가격이 좀 비싸고 발열이 있을 때가 있습니다.
전반적으로 만족스러운 플래그십 스마트폰이에요.
"""
result = analyze_review(review)
print(f"제품: {result.product_name}")
print(f"평점: {result.overall_rating}/5")
print(f"장점: {', '.join(result.pros)}")
print(f"단점: {', '.join(result.cons)}")
print(f"추천: {'예' if result.recommendation else '아니오'}")
Vision API
from pathlib import Path
def encode_image(image_path: str) -> str:
with open(image_path, "rb") as image_file:
return base64.b64encode(image_file.read()).decode("utf-8")
def analyze_image(image_path: str, question: str = "이 이미지를 자세히 설명해주세요.") -> str:
로컬 이미지 분석
base64_image = encode_image(image_path)
ext = Path(image_path).suffix.lower().replace('.', '')
media_type = f"image/{ext if ext in ['jpeg', 'png', 'gif', 'webp'] else 'jpeg'}"
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {
"url": f"data:{media_type};base64,{base64_image}",
"detail": "high" # "low" 또는 "high"
}
},
{
"type": "text",
"text": question
}
]
}
],
max_tokens=1024,
)
return response.choices[0].message.content
URL 이미지 분석
def analyze_image_url(url: str, question: str) -> str:
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "user",
"content": [
{"type": "image_url", "image_url": {"url": url}},
{"type": "text", "text": question}
]
}
],
)
return response.choices[0].message.content
임베딩 API
def get_embeddings(texts: list, model: str = "text-embedding-3-small") -> list:
response = client.embeddings.create(
input=texts,
model=model,
)
return [item.embedding for item in response.data]
의미론적 검색
from sklearn.metrics.pairwise import cosine_similarity
class SemanticSearch:
def __init__(self):
self.documents = []
self.embeddings = None
def index(self, documents: list):
self.documents = documents
self.embeddings = np.array(get_embeddings(documents))
def search(self, query: str, top_k: int = 3) -> list:
query_emb = np.array(get_embeddings([query]))
sims = cosine_similarity(query_emb, self.embeddings)[0]
top_idx = np.argsort(sims)[::-1][:top_k]
return [(self.documents[i], float(sims[i])) for i in top_idx]
3. Anthropic Claude API
기본 설정
pip install anthropic
client = anthropic.Anthropic(
api_key=os.environ.get("ANTHROPIC_API_KEY"),
)
Messages API
def claude_chat(
user_message: str,
system: str = "You are a helpful assistant.",
model: str = "claude-3-5-sonnet-20241022",
max_tokens: int = 1024,
) -> str:
message = client.messages.create(
model=model,
max_tokens=max_tokens,
system=system,
messages=[
{"role": "user", "content": user_message}
]
)
return message.content[0].text
멀티턴 대화
def claude_conversation(messages: list, system: str = None) -> str:
kwargs = {
"model": "claude-3-5-sonnet-20241022",
"max_tokens": 2048,
"messages": messages
}
if system:
kwargs["system"] = system
message = client.messages.create(**kwargs)
return message.content[0].text
스트리밍
def claude_stream(user_message: str, system: str = None) -> str:
full_text = ""
kwargs = {
"model": "claude-3-5-sonnet-20241022",
"max_tokens": 2048,
"messages": [{"role": "user", "content": user_message}]
}
if system:
kwargs["system"] = system
with client.messages.stream(**kwargs) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
full_text += text
print()
return full_text
System Prompt 설계 모범 사례
좋은 시스템 프롬프트 구성 요소
system_prompt_template = """
당신은 [역할]입니다.
주요 역할
- [역할 1]
- [역할 2]
행동 지침
1. [지침 1]
2. [지침 2]
제약 사항
- [제약 1]
- [제약 2]
응답 형식
- [형식 지침]
"""
한국어 고객 지원 봇 예시
cs_system_prompt = """
당신은 한국의 전자상거래 플랫폼 "쇼핑몰"의 고객 지원 AI 상담원입니다.
주요 역할
- 주문 상태 조회 및 안내
- 환불 및 교환 정책 안내
- 배송 관련 문의 처리
- 제품 정보 제공
행동 지침
1. 항상 정중하고 친절한 말투를 사용하세요.
2. 질문에 대한 답을 모를 경우 솔직히 인정하고 전문 상담원 연결을 안내하세요.
3. 개인정보(주소, 카드번호 등)를 요청하지 마세요.
4. 경쟁사 비교 질문에는 중립적으로 답변하세요.
제약 사항
- 회사 내부 정보나 직원 개인 정보를 공개하지 마세요.
- 법적 조언이나 의료 정보를 제공하지 마세요.
- 확인되지 않은 프로모션이나 할인 정보를 약속하지 마세요.
응답 형식
- 간결하고 명확하게 답변하세요.
- 필요시 번호 목록을 사용해 단계별로 안내하세요.
- 마지막에 추가 도움이 필요한지 물어보세요.
"""
Extended Thinking (Claude 3.7+)
def claude_think(question: str, budget_tokens: int = 8000) -> dict:
"""Extended Thinking으로 복잡한 추론 문제 해결"""
response = client.messages.create(
model="claude-3-7-sonnet-20250219",
max_tokens=16000,
thinking={
"type": "enabled",
"budget_tokens": budget_tokens # 내부 추론에 사용할 최대 토큰
},
messages=[{"role": "user", "content": question}]
)
result = {"thinking": "", "answer": ""}
for block in response.content:
if block.type == "thinking":
result["thinking"] = block.thinking
elif block.type == "text":
result["answer"] = block.text
return result
복잡한 수학 문제
problem = """
한 회사의 연간 매출이 매년 15% 씩 증가합니다.
현재 매출이 100억 원이라면, 5년 후 매출이 200억 원을 넘는지 계산하고
정확한 5년 후 매출액도 계산해주세요.
"""
result = claude_think(problem)
print("추론 과정 (일부):", result["thinking"][:300])
print("\n최종 답변:", result["answer"])
Tool Use (Claude)
client = anthropic.Anthropic()
도구 정의
tools = [
{
"name": "calculate",
"description": "수학 계산을 수행합니다.",
"input_schema": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "계산할 수식 (예: 2 + 2, 15 * 30)"
}
},
"required": ["expression"]
}
}
]
def process_tool_call(tool_name: str, tool_input: dict) -> str:
if tool_name == "calculate":
try:
실제 프로덕션에서는 eval 대신 안전한 계산기 사용
result = eval(tool_input["expression"])
return str(result)
except Exception as e:
return f"오류: {str(e)}"
return "알 수 없는 도구"
def claude_with_tools(user_message: str) -> str:
messages = [{"role": "user", "content": user_message}]
while True:
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
tools=tools,
messages=messages
)
도구 호출 없으면 최종 응답 반환
if response.stop_reason == "end_turn":
return response.content[0].text
도구 호출 처리
tool_results = []
for content_block in response.content:
if content_block.type == "tool_use":
result = process_tool_call(content_block.name, content_block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": content_block.id,
"content": result
})
메시지 기록 업데이트
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
Prompt Caching (비용 절감)
긴 시스템 프롬프트나 문서를 캐시해 비용 절감
첫 호출 후 동일한 프롬프트는 90% 할인
long_document = "..." * 1000 # 긴 문서
def analyze_document_with_cache(question: str) -> str:
response = client.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1024,
system=[
{
"type": "text",
"text": "다음 문서를 분석하고 질문에 답변하세요:",
"cache_control": {"type": "ephemeral"} # 캐시 지시자
},
{
"type": "text",
"text": long_document,
"cache_control": {"type": "ephemeral"}
}
],
messages=[{"role": "user", "content": question}]
)
캐시 사용 여부 확인
usage = response.usage
print(f"입력 토큰: {usage.input_tokens}")
print(f"캐시 생성 토큰: {getattr(usage, 'cache_creation_input_tokens', 0)}")
print(f"캐시 읽기 토큰: {getattr(usage, 'cache_read_input_tokens', 0)}")
return response.content[0].text
4. Google Gemini API
기본 설정
pip install google-generativeai
genai.configure(api_key=os.environ.get("GOOGLE_API_KEY"))
모델 초기화
model = genai.GenerativeModel(
model_name="gemini-1.5-flash",
generation_config=genai.GenerationConfig(
temperature=0.7,
top_p=0.9,
top_k=40,
max_output_tokens=2048,
),
system_instruction="당신은 도움이 되는 한국어 AI 어시스턴트입니다."
)
기본 사용
텍스트 생성
response = model.generate_content("Python의 비동기 프로그래밍을 설명해주세요.")
print(response.text)
채팅 세션
chat = model.start_chat(history=[])
response1 = chat.send_message("안녕하세요! 머신러닝에 대해 알고 싶어요.")
print(response1.text)
response2 = chat.send_message("그럼 딥러닝과의 차이점은 무엇인가요?")
print(response2.text)
대화 기록 확인
for turn in chat.history:
print(f"{turn.role}: {turn.parts[0].text[:100]}...")
멀티모달 (텍스트 + 이미지)
from io import BytesIO
model_vision = genai.GenerativeModel("gemini-1.5-pro")
로컬 이미지
image = PIL.Image.open("chart.png")
response = model_vision.generate_content([
image,
"이 차트를 분석하고 주요 트렌드를 한국어로 설명해주세요."
])
print(response.text)
URL 이미지
response_url = requests.get("https://example.com/image.jpg")
image_from_url = PIL.Image.open(BytesIO(response_url.content))
response = model_vision.generate_content([
image_from_url,
"이 이미지에서 텍스트를 추출해주세요."
])
PDF 처리 (Gemini 1.5의 강점)
with open("report.pdf", "rb") as f:
pdf_data = f.read()
response = model_vision.generate_content([
{"mime_type": "application/pdf", "data": pdf_data},
"이 PDF 문서의 핵심 내용을 요약해주세요."
])
print(response.text)
1M 컨텍스트 윈도우 활용
Gemini 1.5 Pro: 최대 1,000,000 토큰 컨텍스트
코드베이스 전체를 한 번에 분석 가능
def analyze_codebase(code_files: dict) -> str:
"""여러 파일로 구성된 코드베이스 분석"""
model_pro = genai.GenerativeModel("gemini-1.5-pro")
content_parts = ["다음 코드베이스를 분석해주세요:\n\n"]
for filename, code in code_files.items():
content_parts.append(f"파일: {filename}\n```\n{code}\n```\n\n")
content_parts.append("아키텍처, 잠재적 버그, 개선 사항을 알려주세요.")
response = model_pro.generate_content(content_parts)
return response.text
긴 문서 요약
def summarize_long_document(document: str) -> str:
model_flash = genai.GenerativeModel("gemini-1.5-flash")
Flash 모델도 1M 토큰 지원, 더 빠르고 저렴
response = model_flash.generate_content(
f"다음 문서를 5개 핵심 포인트로 요약하세요:\n\n{document}"
)
return response.text
구조화된 출력 (JSON Mode)
def extract_structured_data(text: str, schema: dict) -> dict:
model = genai.GenerativeModel(
"gemini-1.5-flash",
generation_config=genai.GenerationConfig(
response_mime_type="application/json",
response_schema=schema,
)
)
response = model.generate_content(text)
return json.loads(response.text)
예시: 이력서에서 정보 추출
resume_schema = {
"type": "object",
"properties": {
"name": {"type": "string"},
"email": {"type": "string"},
"skills": {"type": "array", "items": {"type": "string"}},
"experience_years": {"type": "number"},
}
}
resume_text = """
홍길동
이메일: hong@example.com
경력: Python 5년, JavaScript 3년, Docker 2년
총 경력: 6년
"""
result = extract_structured_data(
f"이 이력서에서 정보를 추출하세요:\n{resume_text}",
resume_schema
)
print(result)
5. 스트리밍 챗봇 구현
FastAPI 백엔드
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from openai import AsyncOpenAI
app = FastAPI()
client = AsyncOpenAI()
class ChatRequest(BaseModel):
message: str
conversation_id: str = None
model: str = "gpt-4o-mini"
대화 기록 저장 (실제로는 Redis나 DB 사용)
conversations = {}
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
conv_id = request.conversation_id or str(time.time())
대화 기록 가져오기
history = conversations.get(conv_id, [])
history.append({"role": "user", "content": request.message})
async def generate():
full_response = ""
try:
stream = await client.chat.completions.create(
model=request.model,
messages=[
{"role": "system", "content": "당신은 도움이 되는 AI 어시스턴트입니다."},
*history
],
stream=True,
max_tokens=2048,
)
async for chunk in stream:
if chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
full_response += content
data = json.dumps({
"content": content,
"conversation_id": conv_id,
"done": False
}, ensure_ascii=False)
yield f"data: {data}\n\n"
대화 기록 업데이트
history.append({"role": "assistant", "content": full_response})
conversations[conv_id] = history[-20:] # 최근 20개만 유지
완료 신호
done_data = json.dumps({
"content": "",
"conversation_id": conv_id,
"done": True
})
yield f"data: {done_data}\n\n"
except Exception as e:
error_data = json.dumps({"error": str(e), "done": True})
yield f"data: {error_data}\n\n"
return StreamingResponse(
generate(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no",
}
)
@app.get("/chat/history/{conversation_id}")
async def get_history(conversation_id: str):
history = conversations.get(conversation_id, [])
return {"conversation_id": conversation_id, "messages": history}
WebSocket 챗봇
from fastapi import WebSocket, WebSocketDisconnect
@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
await websocket.accept()
conversation_history = []
try:
while True:
메시지 수신
data = await websocket.receive_json()
user_message = data.get("message", "")
if not user_message:
continue
conversation_history.append({
"role": "user",
"content": user_message
})
스트리밍 응답 전송
full_response = ""
stream = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "당신은 도움이 되는 AI 어시스턴트입니다."},
*conversation_history[-10:]
],
stream=True,
max_tokens=1024,
)
async for chunk in stream:
if chunk.choices[0].delta.content:
content = chunk.choices[0].delta.content
full_response += content
await websocket.send_json({
"type": "stream",
"content": content
})
완료 신호
await websocket.send_json({"type": "done", "content": full_response})
conversation_history.append({
"role": "assistant",
"content": full_response
})
except WebSocketDisconnect:
print("클라이언트 연결 끊김")
except Exception as e:
await websocket.send_json({"type": "error", "message": str(e)})
6. 대화 메모리 관리
요약 메모리
from openai import OpenAI
client = OpenAI()
class SummaryMemory:
def __init__(self, model: str = "gpt-4o-mini", max_tokens: int = 3000):
self.model = model
self.max_tokens = max_tokens
self.summary = ""
self.recent_messages = []
self.encoder = tiktoken.encoding_for_model(model)
def count_tokens(self, text: str) -> int:
return len(self.encoder.encode(text))
def add_message(self, role: str, content: str):
self.recent_messages.append({"role": role, "content": content})
total_tokens = sum(self.count_tokens(m['content']) for m in self.recent_messages)
if total_tokens > self.max_tokens:
self._summarize_old_messages()
def _summarize_old_messages(self):
절반의 메시지를 요약
messages_to_summarize = self.recent_messages[:len(self.recent_messages)//2]
self.recent_messages = self.recent_messages[len(self.recent_messages)//2:]
conversation_text = "\n".join(
f"{m['role']}: {m['content']}"
for m in messages_to_summarize
)
summary_prompt = f"""
기존 요약: {self.summary}
새 대화:
{conversation_text}
위 대화를 간결하게 요약하세요. 중요한 정보, 결정 사항, 사용자 선호도를 포함하세요.
"""
response = client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": summary_prompt}],
max_tokens=500,
)
self.summary = response.choices[0].message.content
def get_messages_with_context(self) -> list:
messages = []
if self.summary:
messages.append({
"role": "system",
"content": f"이전 대화 요약:\n{self.summary}"
})
messages.extend(self.recent_messages)
return messages
사용 예시
memory = SummaryMemory()
def chat_with_memory(user_input: str) -> str:
memory.add_message("user", user_input)
messages_with_context = memory.get_messages_with_context()
messages_with_context.insert(0, {
"role": "system",
"content": "당신은 도움이 되는 AI 어시스턴트입니다."
})
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages_with_context,
max_tokens=1024,
)
assistant_reply = response.choices[0].message.content
memory.add_message("assistant", assistant_reply)
return assistant_reply
7. 비용 최적화 전략
입력 토큰 최적화
def optimize_prompt(prompt: str, max_tokens: int = 1000) -> str:
"""프롬프트 최적화"""
encoder = tiktoken.encoding_for_model("gpt-4o")
tokens = encoder.encode(prompt)
if len(tokens) <= max_tokens:
return prompt
토큰 수 초과 시 잘라내기
truncated_tokens = tokens[:max_tokens]
return encoder.decode(truncated_tokens)
불필요한 공백 제거
def clean_prompt(prompt: str) -> str:
연속 공백 제거
prompt = re.sub(r' +', ' ', prompt)
연속 줄바꿈 최소화
prompt = re.sub(r'\n{3,}', '\n\n', prompt)
return prompt.strip()
문서 요약 후 사용
def compress_document(document: str, ratio: float = 0.3) -> str:
"""문서를 요약해 토큰 수 절감"""
response = client.chat.completions.create(
model="gpt-4o-mini", # 요약에는 저렴한 모델 사용
messages=[
{
"role": "user",
"content": f"다음 문서를 원문의 {ratio*100:.0f}% 분량으로 핵심만 요약하세요:\n\n{document}"
}
],
max_tokens=int(len(document.split()) * ratio * 1.5),
)
return response.choices[0].message.content
모델 라우팅 전략
from enum import Enum
class TaskComplexity(Enum):
SIMPLE = "simple"
MEDIUM = "medium"
COMPLEX = "complex"
def classify_task_complexity(query: str) -> TaskComplexity:
"""쿼리 복잡도에 따라 모델 선택"""
간단한 규칙 기반 분류
simple_keywords = ["안녕", "날씨", "간단한", "번역", "정의"]
complex_keywords = ["분석", "추론", "코드 작성", "논문", "전략"]
query_lower = query.lower()
if any(kw in query_lower for kw in simple_keywords):
return TaskComplexity.SIMPLE
elif any(kw in query_lower for kw in complex_keywords):
return TaskComplexity.COMPLEX
else:
return TaskComplexity.MEDIUM
def get_optimal_model(complexity: TaskComplexity) -> str:
"""복잡도에 따른 최적 모델 선택"""
model_mapping = {
TaskComplexity.SIMPLE: "gpt-4o-mini", # 빠르고 저렴
TaskComplexity.MEDIUM: "gpt-4o-mini", # 균형적
TaskComplexity.COMPLEX: "gpt-4o", # 고성능
}
return model_mapping[complexity]
def smart_chat(query: str) -> str:
"""복잡도에 따른 스마트 라우팅"""
complexity = classify_task_complexity(query)
model = get_optimal_model(complexity)
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": query}],
max_tokens=1024,
)
print(f"사용 모델: {model} (복잡도: {complexity.value})")
return response.choices[0].message.content
배치 API 활용
def batch_process_documents(documents: list, task: str) -> list:
"""배치 API로 여러 문서 동시 처리 (50% 비용 절감)"""
배치 요청 준비
batch_requests = []
for i, doc in enumerate(documents):
batch_requests.append({
"custom_id": f"request-{i}",
"method": "POST",
"url": "/v1/chat/completions",
"body": {
"model": "gpt-4o-mini",
"messages": [
{"role": "user", "content": f"{task}\n\n{doc}"}
],
"max_tokens": 512
}
})
JSONL 파일 생성
with open("batch_requests.jsonl", "w", encoding="utf-8") as f:
for request in batch_requests:
f.write(json.dumps(request, ensure_ascii=False) + "\n")
배치 파일 업로드
with open("batch_requests.jsonl", "rb") as f:
batch_file = client.files.create(
file=f,
purpose="batch"
)
배치 작업 생성
batch = client.batches.create(
input_file_id=batch_file.id,
endpoint="/v1/chat/completions",
completion_window="24h"
)
print(f"배치 ID: {batch.id}, 상태: {batch.status}")
return batch.id
8. 프로덕션 베스트 프랙티스
재시도 로직
from openai import OpenAI, RateLimitError, APITimeoutError, APIConnectionError
class RobustLLMClient:
def __init__(self, max_retries: int = 5, base_delay: float = 1.0):
self.client = OpenAI()
self.max_retries = max_retries
self.base_delay = base_delay
def create_with_retry(self, **kwargs) -> any:
"""지수 백오프(Exponential Backoff)를 사용한 재시도"""
last_exception = None
for attempt in range(self.max_retries):
try:
return self.client.chat.completions.create(**kwargs)
except RateLimitError as e:
속도 제한: 더 오래 대기
delay = self.base_delay * (2 ** attempt) + random.uniform(0, 1)
print(f"속도 제한 도달. {delay:.1f}초 후 재시도 ({attempt+1}/{self.max_retries})")
time.sleep(delay)
last_exception = e
except APITimeoutError as e:
타임아웃: 빠르게 재시도
delay = self.base_delay * (1.5 ** attempt)
print(f"타임아웃. {delay:.1f}초 후 재시도 ({attempt+1}/{self.max_retries})")
time.sleep(delay)
last_exception = e
except APIConnectionError as e:
연결 오류: 네트워크 복구 대기
delay = self.base_delay * (2 ** attempt) + random.uniform(1, 3)
print(f"연결 오류. {delay:.1f}초 후 재시도 ({attempt+1}/{self.max_retries})")
time.sleep(delay)
last_exception = e
except Exception as e:
재시도 불가능한 오류 (400, 401 등)
raise e
raise last_exception
사용
robust_client = RobustLLMClient()
response = robust_client.create_with_retry(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "안녕하세요"}],
max_tokens=100,
)
로깅과 모니터링
from functools import wraps
from dataclasses import dataclass
구조화된 로깅 설정
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s %(levelname)s %(message)s'
)
logger = logging.getLogger(__name__)
@dataclass
class LLMCallLog:
request_id: str
model: str
input_tokens: int
output_tokens: int
latency_ms: float
success: bool
error: str = None
estimated_cost_usd: float = 0.0
def log_llm_call(func):
"""LLM API 호출 자동 로깅 데코레이터"""
@wraps(func)
def wrapper(*args, **kwargs):
request_id = str(uuid.uuid4())[:8]
start_time = time.time()
log = LLMCallLog(
request_id=request_id,
model=kwargs.get('model', 'unknown'),
input_tokens=0,
output_tokens=0,
latency_ms=0,
success=False
)
try:
result = func(*args, **kwargs)
log.success = True
log.latency_ms = (time.time() - start_time) * 1000
if hasattr(result, 'usage'):
log.input_tokens = result.usage.prompt_tokens
log.output_tokens = result.usage.completion_tokens
log.estimated_cost_usd = estimate_cost(
log.model,
log.input_tokens,
log.output_tokens
)
logger.info(json.dumps({
"request_id": log.request_id,
"model": log.model,
"input_tokens": log.input_tokens,
"output_tokens": log.output_tokens,
"latency_ms": round(log.latency_ms, 2),
"cost_usd": round(log.estimated_cost_usd, 6),
"success": log.success
}))
return result
except Exception as e:
log.success = False
log.error = str(e)
log.latency_ms = (time.time() - start_time) * 1000
logger.error(json.dumps({
"request_id": log.request_id,
"model": log.model,
"error": log.error,
"latency_ms": round(log.latency_ms, 2),
"success": False
}))
raise
return wrapper
@log_llm_call
def logged_chat(model: str, messages: list, **kwargs) -> any:
return client.chat.completions.create(
model=model,
messages=messages,
**kwargs
)
9. 실전 프로젝트: 코드 리뷰 봇
GitHub 웹훅 설정
from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
from openai import AsyncOpenAI
app = FastAPI()
client = AsyncOpenAI()
GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN")
GITHUB_WEBHOOK_SECRET = os.environ.get("GITHUB_WEBHOOK_SECRET")
def verify_webhook_signature(payload: bytes, signature: str) -> bool:
"""GitHub 웹훅 서명 검증"""
expected = hmac.new(
GITHUB_WEBHOOK_SECRET.encode(),
payload,
hashlib.sha256
).hexdigest()
return hmac.compare_digest(f"sha256={expected}", signature)
async def get_pr_diff(owner: str, repo: str, pr_number: int) -> str:
"""PR의 코드 변경 사항 가져오기"""
async with httpx.AsyncClient() as http_client:
response = await http_client.get(
f"https://api.github.com/repos/{owner}/{repo}/pulls/{pr_number}",
headers={
"Authorization": f"token {GITHUB_TOKEN}",
"Accept": "application/vnd.github.v3.diff"
}
)
return response.text
async def review_code_with_llm(diff: str, pr_title: str) -> str:
"""LLM으로 코드 리뷰 생성"""
system_prompt = """당신은 경험 많은 시니어 개발자입니다.
PR의 코드 변경 사항을 검토하고 건설적인 피드백을 한국어로 제공하세요.
검토 항목:
1. 버그 가능성
2. 성능 이슈
3. 보안 취약점
4. 코드 가독성
5. 모범 사례 준수 여부
마크다운 형식으로 작성하세요."""
user_prompt = f"""PR 제목: {pr_title}
코드 변경 사항(Diff excerpt):
{diff[:8000]}
위 코드 변경 사항을 리뷰해주세요."""
response = await client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
max_tokens=2048,
temperature=0.3,
)
return response.choices[0].message.content
async def post_review_comment(owner: str, repo: str, pr_number: int, review: str):
"""PR에 리뷰 코멘트 작성"""
async with httpx.AsyncClient() as http_client:
await http_client.post(
f"https://api.github.com/repos/{owner}/{repo}/issues/{pr_number}/comments",
json={"body": f"## AI 코드 리뷰\n\n{review}"},
headers={
"Authorization": f"token {GITHUB_TOKEN}",
"Accept": "application/vnd.github.v3+json"
}
)
async def process_pr_event(payload: dict):
"""PR 이벤트 처리"""
action = payload.get("action")
if action not in ["opened", "synchronize", "reopened"]:
return
pr = payload["pull_request"]
owner = payload["repository"]["owner"]["login"]
repo = payload["repository"]["name"]
pr_number = pr["number"]
pr_title = pr["title"]
print(f"PR #{pr_number} 리뷰 시작: {pr_title}")
try:
diff = await get_pr_diff(owner, repo, pr_number)
review = await review_code_with_llm(diff, pr_title)
await post_review_comment(owner, repo, pr_number, review)
print(f"PR #{pr_number} 리뷰 완료")
except Exception as e:
print(f"PR #{pr_number} 리뷰 실패: {e}")
@app.post("/webhook/github")
async def github_webhook(
request: Request,
background_tasks: BackgroundTasks
):
payload_bytes = await request.body()
signature = request.headers.get("X-Hub-Signature-256", "")
if not verify_webhook_signature(payload_bytes, signature):
raise HTTPException(status_code=401, detail="Invalid signature")
event_type = request.headers.get("X-GitHub-Event")
payload = await request.json()
if event_type == "pull_request":
background_tasks.add_task(process_pr_event, payload)
return {"status": "accepted"}
고급 코드 리뷰 기능
from typing import List, Dict
class CodeReviewBot:
def __init__(self):
self.client = AsyncOpenAI()
async def review_file(self, filename: str, content: str, diff: str) -> Dict:
"""파일별 상세 리뷰"""
response = await self.client.chat.completions.create(
model="gpt-4o",
messages=[
{
"role": "system",
"content": """코드 리뷰 전문가입니다. JSON 형식으로 피드백을 제공하세요.
각 이슈에 심각도(critical/major/minor)와 제안 사항을 포함하세요."""
},
{
"role": "user",
"content": f"""
파일명: {filename}
변경 내용(Diff excerpt):
{diff}
현재 파일 전체(Excerpt):
{content[:3000]}
이 파일의 코드를 리뷰하고 JSON 형식으로 반환하세요.
형식: {{"issues": [{{"line": 0, "severity": "critical", "message": "", "suggestion": ""}}], "summary": ""}}
"""
}
],
response_format={"type": "json_object"},
temperature=0,
)
return json.loads(response.choices[0].message.content)
def format_review_comment(self, file_reviews: List[Dict]) -> str:
"""리뷰 결과를 마크다운으로 포매팅"""
comment = "## 자동 코드 리뷰 결과\n\n"
critical_count = 0
major_count = 0
minor_count = 0
for file_review in file_reviews:
filename = file_review.get('filename', 'unknown')
review = file_review.get('review', {})
issues = review.get('issues', [])
for issue in issues:
severity = issue.get('severity', 'minor')
if severity == 'critical':
critical_count += 1
elif severity == 'major':
major_count += 1
else:
minor_count += 1
요약 섹션
comment += f"### 요약\n"
comment += f"- 치명적(Critical): {critical_count}개\n"
comment += f"- 중요(Major): {major_count}개\n"
comment += f"- 경미(Minor): {minor_count}개\n\n"
파일별 상세 내용
for file_review in file_reviews:
filename = file_review.get('filename', 'unknown')
review = file_review.get('review', {})
comment += f"### {filename}\n"
comment += f"{review.get('summary', '')}\n\n"
issues = review.get('issues', [])
if issues:
comment += "**이슈:**\n"
severity_emoji = {'critical': '🔴', 'major': '🟡', 'minor': '🟢'}
for issue in issues:
emoji = severity_emoji.get(issue.get('severity', 'minor'), '⚪')
comment += f"- {emoji} 라인 {issue.get('line', '?')}: {issue.get('message', '')}\n"
if issue.get('suggestion'):
comment += f" - 제안: {issue['suggestion']}\n"
comment += "\n"
return comment
마무리
LLM API를 활용한 애플리케이션 개발에서 핵심 원칙들:
**선택 기준**:
- 범용 챗봇: GPT-4o-mini (비용 효율) 또는 Claude 3.5 Sonnet (품질)
- 코딩/추론: GPT-4o 또는 Claude 3.7 (Extended Thinking)
- 긴 문서 처리: Gemini 1.5 Pro (1M 컨텍스트)
- 비용 최소화: Gemini 1.5 Flash 또는 gpt-4o-mini
**프로덕션 필수 요소**:
- 재시도 로직 (지수 백오프)
- 속도 제한 처리
- 구조화된 로깅
- 비용 모니터링
- 스트리밍으로 UX 개선
**비용 절감**:
- 작은 태스크에는 소형 모델 사용
- 배치 API 활용 (50% 절감)
- Anthropic Prompt Cache 활용 (90% 절감)
- 프롬프트 최적화
LLM API 개발에서 가장 중요한 것은 "올바른 도구를 올바른 용도로 사용하는 것"입니다. 모든 태스크에 GPT-4o를 쓸 필요가 없고, 적절한 모델 라우팅과 캐싱 전략으로 비용을 크게 줄일 수 있습니다.
현재 단락 (1/1066)
LLM API를 처음 사용하면 "Hello World"는 금방 만들 수 있지만, 실제 프로덕션 서비스를 개발할 때는 수많은 함정이 있습니다. 스트리밍 구현, 비용 폭탄 방지, 속도...