OpenAI AI Deployment Engineer (Seoul) 합격 가이드: GPT 엔터프라이즈 배포 전문가가 되는 완벽 로드맵

import asyncio
import aiohttp
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def process_documents(documents: list[str]) -> list[str]:
    """여러 문서를 병렬로 요약합니다."""
    tasks = [summarize(doc) for doc in documents]
    return await asyncio.gather(*tasks)

async def summarize(document: str) -> str:
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "문서를 3줄로 요약하세요."},
            {"role": "user", "content": document}
        ],
        temperature=0.3
    )
    return response.choices[0].message.content

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from openai import OpenAI

app = FastAPI(title="Enterprise AI API")
client = OpenAI()

class QueryRequest(BaseModel):
    question: str = Field(..., min_length=1, max_length=2000)
    context: str | None = None
    model: str = "gpt-4o"

class QueryResponse(BaseModel):
    answer: str
    tokens_used: int
    model: str

@app.post("/query", response_model=QueryResponse)
async def query_ai(request: QueryRequest):
    try:
        response = client.chat.completions.create(
            model=request.model,
            messages=[
                {"role": "system", "content": "당신은 도움이 되는 AI 어시스턴트입니다."},
                {"role": "user", "content": request.question}
            ]
        )
        return QueryResponse(
            answer=response.choices[0].message.content,
            tokens_used=response.usage.total_tokens,
            model=request.model
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

from pydantic import BaseModel, Field
from enum import Enum

class Sentiment(str, Enum):
    POSITIVE = "positive"
    NEGATIVE = "negative"
    NEUTRAL = "neutral"

class ReviewAnalysis(BaseModel):
    sentiment: Sentiment
    confidence: float = Field(ge=0.0, le=1.0)
    key_topics: list[str] = Field(max_length=5)
    summary: str = Field(max_length=200)

# Chain of Thought (CoT) 프롬프팅
system_prompt = """
당신은 금융 리스크 분석 전문가입니다.
문제를 분석할 때 다음 단계를 따르세요:
1. 핵심 리스크 요인을 식별합니다
2. 각 요인의 영향도를 평가합니다
3. 상관관계를 분석합니다
4. 최종 리스크 등급을 결정합니다
각 단계의 추론 과정을 명시적으로 보여주세요.
"""

# Few-shot 프롬프팅
few_shot_prompt = """
다음 고객 리뷰의 감정을 분석하세요.

리뷰: "배송은 빨랐지만 제품 품질이 기대 이하였습니다."
분석: 혼합 (배송 긍정, 품질 부정), 전체 감정: 부정

리뷰: "가격 대비 성능이 훌륭하고 A/S도 만족합니다."
분석: 긍정 (가격, 성능, A/S 모두 긍정), 전체 감정: 긍정

리뷰: "{input_review}"
분석:
"""

from openai import OpenAI
import json

client = OpenAI()

# 1. 학습 데이터 준비 (JSONL 포맷)
training_data = [
    {
        "messages": [
            {"role": "system", "content": "당신은 한국 금융 상품 전문가입니다."},
            {"role": "user", "content": "ISA 계좌의 세제 혜택은?"},
            {"role": "assistant", "content": "ISA(Individual Savings Account)..."}
        ]
    }
    # ... 수백-수천 건의 학습 데이터
]

# 2. 파일 업로드
file = client.files.create(
    file=open("training_data.jsonl", "rb"),
    purpose="fine-tune"
)

# 3. 파인튜닝 작업 생성
job = client.fine_tuning.jobs.create(
    training_file=file.id,
    model="gpt-4o-mini-2024-07-18",
    hyperparameters={
        "n_epochs": 3,
        "batch_size": "auto",
        "learning_rate_multiplier": "auto"
    }
)

# 4. 파인튜닝된 모델 사용
response = client.chat.completions.create(
    model=job.fine_tuned_model,
    messages=[
        {"role": "user", "content": "퇴직연금 IRP의 세액공제 한도는?"}
    ]
)

from openai import OpenAI
import numpy as np

client = OpenAI()

def get_embedding(text: str, model: str = "text-embedding-3-large") -> list[float]:
    response = client.embeddings.create(
        input=text,
        model=model,
        dimensions=1536  # 차원 축소 가능
    )
    return response.data[0].embedding

def cosine_similarity(a: list[float], b: list[float]) -> float:
    a_np, b_np = np.array(a), np.array(b)
    return np.dot(a_np, b_np) / (np.linalg.norm(a_np) * np.linalg.norm(b_np))

tools = [
    {
        "type": "function",
        "function": {
            "name": "search_knowledge_base",
            "description": "회사 내부 지식 베이스를 검색합니다",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "검색 쿼리"
                    },
                    "department": {
                        "type": "string",
                        "enum": ["HR", "Engineering", "Finance", "Legal"]
                    }
                },
                "required": ["query"]
            }
        }
    }
]

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

from pydantic import BaseModel

class ExtractedEntity(BaseModel):
    name: str
    entity_type: str
    confidence: float

class ExtractionResult(BaseModel):
    entities: list[ExtractedEntity]
    raw_text: str

response = client.beta.chat.completions.parse(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "텍스트에서 엔티티를 추출하세요."},
        {"role": "user", "content": "삼성전자가 AI 반도체 투자를 50조원으로 확대합니다."}
    ],
    response_format=ExtractionResult
)

result = response.choices[0].message.parsed

문서 로딩 -> 청킹 -> 임베딩 -> 벡터 저장 -> 검색 -> 리랭킹 -> 생성

from openai import OpenAI
from pinecone import Pinecone
from typing import List

client = OpenAI()
pc = Pinecone(api_key="your-key")
index = pc.Index("enterprise-docs")

def chunk_document(text: str, chunk_size: int = 500, overlap: int = 100) -> list[str]:
    """문서를 오버래핑 청크로 분할합니다."""
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        chunks.append(chunk)
        start += chunk_size - overlap
    return chunks

def index_document(doc_id: str, text: str):
    """문서를 벡터 DB에 인덱싱합니다."""
    chunks = chunk_document(text)
    for i, chunk in enumerate(chunks):
        embedding = get_embedding(chunk)
        index.upsert(vectors=[{
            "id": f"doc-{doc_id}-chunk-{i}",
            "values": embedding,
            "metadata": {"text": chunk, "doc_id": doc_id, "chunk_index": i}
        }])

def retrieve_and_generate(query: str, top_k: int = 5) -> str:
    """검색 증강 생성을 수행합니다."""
    query_embedding = get_embedding(query)
    results = index.query(vector=query_embedding, top_k=top_k, include_metadata=True)

    context = "\n\n".join([match.metadata["text"] for match in results.matches])

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": (
                    "다음 컨텍스트를 기반으로 질문에 답하세요. "
                    "컨텍스트에 없는 정보는 '해당 정보를 찾을 수 없습니다'라고 답하세요."
                )
            },
            {"role": "user", "content": f"컨텍스트:\n{context}\n\n질문: {query}"}
        ],
        temperature=0.1
    )
    return response.choices[0].message.content

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_recall,
    context_precision
)

# 평가 데이터셋 준비
eval_dataset = {
    "question": ["ISA 계좌의 비과세 한도는?"],
    "answer": ["ISA 계좌의 비과세 한도는 200만원입니다."],
    "contexts": [["ISA 일반형의 비과세 한도는 200만원이며..."]],
    "ground_truth": ["ISA 일반형 200만원, 서민형 400만원"]
}

result = evaluate(
    dataset=eval_dataset,
    metrics=[faithfulness, answer_relevancy, context_recall, context_precision]
)

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    current_step: str
    findings: list[str]

def research_node(state: AgentState) -> AgentState:
    """정보를 조사합니다."""
    # 도구를 사용해 관련 정보 검색
    return {"messages": [{"role": "assistant", "content": "조사 완료"}],
            "findings": ["finding1"]}

def analyze_node(state: AgentState) -> AgentState:
    """조사 결과를 분석합니다."""
    return {"messages": [{"role": "assistant", "content": "분석 완료"}]}

def should_continue(state: AgentState) -> str:
    """추가 조사가 필요한지 판단합니다."""
    if len(state["findings"]) < 3:
        return "research"
    return "end"

# 그래프 구성
workflow = StateGraph(AgentState)
workflow.add_node("research", research_node)
workflow.add_node("analyze", analyze_node)
workflow.add_edge("research", "analyze")
workflow.add_conditional_edges("analyze", should_continue,
    {"research": "research", "end": END})
workflow.set_entry_point("research")

app = workflow.compile()

from crewai import Agent, Task, Crew

researcher = Agent(
    role="시장 분석가",
    goal="산업 트렌드와 경쟁사 분석",
    backstory="10년 경력의 시장 분석 전문가"
)

writer = Agent(
    role="보고서 작성자",
    goal="분석 결과를 경영진용 보고서로 작성",
    backstory="컨설팅 펌 출신의 보고서 전문가"
)

research_task = Task(
    description="한국 AI 시장의 2026년 전망을 분석하세요",
    agent=researcher
)

write_task = Task(
    description="분석 결과를 경영진 보고서로 작성하세요",
    agent=writer
)

crew = Crew(agents=[researcher, writer], tasks=[research_task, write_task])
result = crew.kickoff()

# 어시스턴트 생성
assistant = client.beta.assistants.create(
    name="Enterprise Data Analyst",
    instructions="기업 데이터를 분석하고 인사이트를 제공합니다.",
    model="gpt-4o",
    tools=[
        {"type": "code_interpreter"},
        {"type": "file_search"}
    ]
)

# LangSmith 트레이싱 예시
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-key"

# Helicone 프록시를 통한 모니터링
from openai import OpenAI

client = OpenAI(
    base_url="https://oai.helicone.ai/v1",
    default_headers={
        "Helicone-Auth": "Bearer your-key"
    }
)

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

apiVersion: apps/v1
kind: Deployment
metadata:
  name: ai-api-server
spec:
  replicas: 3
  selector:
    matchLabels:
      app: ai-api
  template:
    metadata:
      labels:
        app: ai-api
    spec:
      containers:
        - name: ai-api
          image: your-registry/ai-api:v1.0
          ports:
            - containerPort: 8000
          env:
            - name: OPENAI_API_KEY
              valueFrom:
                secretKeyRef:
                  name: openai-secrets
                  key: api-key
          resources:
            requests:
              cpu: '500m'
              memory: '512Mi'
            limits:
              cpu: '1000m'
              memory: '1Gi'
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: ai-api-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: ai-api-server
  minReplicas: 2
  maxReplicas: 10
  metrics:
    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 70

from openai import AzureOpenAI

client = AzureOpenAI(
    azure_endpoint="https://your-resource.openai.azure.com/",
    api_key="your-azure-key",
    api_version="2024-06-01"
)

response = client.chat.completions.create(
    model="gpt-4o-deployment-name",
    messages=[{"role": "user", "content": "안녕하세요"}]
)

-- 고객별 최근 3건의 AI 쿼리 분석
WITH ranked_queries AS (
    SELECT
        customer_id,
        query_text,
        response_quality_score,
        token_count,
        created_at,
        ROW_NUMBER() OVER (
            PARTITION BY customer_id
            ORDER BY created_at DESC
        ) as rn,
        AVG(response_quality_score) OVER (
            PARTITION BY customer_id
            ORDER BY created_at
            ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
        ) as rolling_avg_quality
    FROM ai_query_logs
)
SELECT * FROM ranked_queries WHERE rn <= 3;

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def extract_documents():
    """고객 문서를 추출합니다."""
    pass

def process_embeddings():
    """임베딩을 생성합니다."""
    pass

def update_vector_store():
    """벡터 스토어를 업데이트합니다."""
    pass

with DAG(
    dag_id="rag_pipeline",
    schedule_interval="@daily",
    start_date=datetime(2026, 1, 1),
    catchup=False
) as dag:
    extract = PythonOperator(task_id="extract", python_callable=extract_documents)
    embed = PythonOperator(task_id="embed", python_callable=process_embeddings)
    update = PythonOperator(task_id="update", python_callable=update_vector_store)

    extract >> embed >> update

// app/api/chat/route.ts
import OpenAI from 'openai'
import { NextResponse } from 'next/server'

const openai = new OpenAI()

export async function POST(request: Request) {
  const { messages } = await request.json()

  const stream = await openai.chat.completions.create({
    model: 'gpt-4o',
    messages,
    stream: true,
  })

  const encoder = new TextEncoder()
  const readable = new ReadableStream({
    async start(controller) {
      for await (const chunk of stream) {
        const text = chunk.choices[0]?.delta?.content || ''
        controller.enqueue(encoder.encode(text))
      }
      controller.close()
    },
  })

  return new Response(readable, {
    headers: { 'Content-Type': 'text/plain; charset=utf-8' },
  })
}

1. 요구사항 명확화 (5분)
   - 기능 요구사항
   - 비기능 요구사항 (QPS, 레이턴시, 가용성)
   - 제약사항 (예산, 기존 인프라)

2. 하이레벨 아키텍처 (10분)
   - 핵심 컴포넌트 식별
   - 데이터 흐름 설계
   - 기술 선택 이유

3. 딥다이브 (15분)
   - 가장 복잡한 컴포넌트 상세 설계
   - 엣지 케이스 처리
   - 확장성 고려

4. 트레이드오프 논의 (5분)
   - 비용 vs 성능
   - 정확도 vs 레이턴시
   - 복잡성 vs 유지보수성