AI 시대 생존 가이드 2편: 개발자에서 LLM 엔지니어/AI 엔지니어로 전환하기

# OpenAI API 기본 사용
from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "당신은 도움이 되는 코드 리뷰어입니다."},
        {"role": "user", "content": "이 Python 함수를 리뷰해줘: def add(a, b): return a + b"}
    ],
    temperature=0.7,
    max_tokens=500
)

print(response.choices[0].message.content)
print(f"사용된 토큰: {response.usage.total_tokens}")
print(f"비용 (gpt-4o 기준): ${response.usage.total_tokens * 0.000005:.6f}")

# 구조화된 출력 예시
from openai import OpenAI
from pydantic import BaseModel
from typing import List

class CodeReview(BaseModel):
    issues: List[str]
    suggestions: List[str]
    severity: str  # "low", "medium", "high"
    score: int     # 1-10

client = OpenAI()

response = client.beta.chat.completions.parse(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "당신은 시니어 개발자입니다. 코드를 리뷰하고 JSON으로 결과를 반환하세요."},
        {"role": "user", "content": "다음 코드를 리뷰해주세요: [코드 내용]"}
    ],
    response_format=CodeReview
)

review = response.choices[0].message.parsed
print(f"심각도: {review.severity}")
print(f"점수: {review.score}/10")

# LangChain으로 RAG 시스템 구축
from langchain_community.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA

# 1. 문서 로드
loader = PyPDFLoader("company_docs.pdf")
documents = loader.load()

# 2. 청킹
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
chunks = text_splitter.split_documents(documents)

# 3. 임베딩 및 벡터 DB 저장
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 4. RAG 체인 생성
llm = ChatOpenAI(model="gpt-4o", temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5})
)

# 5. 질문
result = qa_chain.invoke({"query": "우리 회사의 연차 정책이 어떻게 되나요?"})
print(result["result"])

# LangChain Agent 예시: 코드 리뷰 에이전트
from langchain import hub
from langchain.agents import create_react_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain_community.utilities import SerpAPIWrapper

# 도구 정의
def analyze_code_complexity(code: str) -> str:
    """코드의 복잡도를 분석합니다."""
    # 실제 구현에서는 radon 등 라이브러리 사용
    lines = code.count('\n')
    functions = code.count('def ')
    return f"코드 줄 수: {lines}, 함수 수: {functions}, 복잡도 점수: {lines // 10}"

def check_security_issues(code: str) -> str:
    """기본적인 보안 취약점을 검사합니다."""
    issues = []
    if 'eval(' in code:
        issues.append("eval() 사용 발견 - 보안 위험")
    if 'shell=True' in code:
        issues.append("shell=True 사용 발견 - 명령 인젝션 위험")
    if not issues:
        return "명백한 보안 취약점 없음"
    return "\n".join(issues)

tools = [
    Tool(
        name="코드_복잡도_분석",
        func=analyze_code_complexity,
        description="Python 코드의 복잡도를 분석합니다. 입력: 코드 문자열"
    ),
    Tool(
        name="보안_취약점_검사",
        func=check_security_issues,
        description="코드에서 보안 취약점을 검사합니다. 입력: 코드 문자열"
    )
]

llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = hub.pull("hwchase17/react")

agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = agent_executor.invoke({
    "input": "다음 코드를 종합적으로 리뷰해주세요: [코드 내용]"
})

from crewai import Agent, Task, Crew

# 에이전트 정의
researcher = Agent(
    role="기술 조사원",
    goal="주어진 주제에 대한 최신 기술 동향 조사",
    backstory="10년 경력의 기술 분석가",
    verbose=True
)

writer = Agent(
    role="기술 블로그 작가",
    goal="조사 결과를 이해하기 쉬운 블로그 포스트로 작성",
    backstory="개발자를 위한 기술 블로그 전문 작가",
    verbose=True
)

# 태스크 정의
research_task = Task(
    description="LLM 엔지니어링 최신 트렌드 조사",
    agent=researcher
)

writing_task = Task(
    description="조사 결과를 바탕으로 블로그 포스트 작성",
    agent=writer
)

# 크루 실행
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task]
)
result = crew.kickoff()

from transformers import AutoModelForCausalLM, AutoTokenizer
from peft import LoraConfig, get_peft_model, TaskType
from trl import SFTTrainer
from datasets import load_dataset

# 기본 모델 로드
model_name = "beomi/Llama-3-Open-Ko-8B"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    load_in_4bit=True,  # QLoRA: 4비트 양자화
    device_map="auto"
)

# LoRA 설정
lora_config = LoraConfig(
    r=16,           # 랭크 (낮을수록 파라미터 수 적음)
    lora_alpha=32,  # 스케일 파라미터
    target_modules=["q_proj", "v_proj"],  # 어텐션 레이어에만 적용
    lora_dropout=0.05,
    bias="none",
    task_type=TaskType.CAUSAL_LM
)

model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# 출력 예: trainable params: 3,407,872 (전체의 0.04%)

from ragas import evaluate
from ragas.metrics import (
    faithfulness,        # 답변이 컨텍스트에 충실한지
    answer_relevancy,    # 답변이 질문에 관련있는지
    context_precision,   # 검색된 컨텍스트가 정확한지
    context_recall       # 필요한 컨텍스트가 모두 검색됐는지
)

# 평가 데이터셋
from datasets import Dataset

data = {
    "question": ["회사의 연차 정책은?", "급여는 언제 지급되나요?"],
    "answer": ["연차는 15일 제공됩니다.", "매월 25일에 지급됩니다."],
    "contexts": [
        ["연차는 연 15일 제공되며 잔여 연차는 이월 불가..."],
        ["급여 지급일은 매월 25일이며 공휴일인 경우 전일 지급..."]
    ],
    "ground_truth": ["연 15일", "매월 25일"]
}

dataset = Dataset.from_dict(data)
result = evaluate(dataset, metrics=[faithfulness, answer_relevancy])
print(result)

# vLLM 설치 및 모델 서빙
pip install vllm

# API 서버 시작
python -m vllm.entrypoints.openai.api_server \
  --model meta-llama/Llama-3-8B-Instruct \
  --max-model-len 4096 \
  --gpu-memory-utilization 0.9

# 비용 모니터링 래퍼 예시
import time
from functools import wraps
from dataclasses import dataclass

COST_PER_TOKEN = {
    "gpt-4o": {"input": 0.000005, "output": 0.000015},
    "gpt-4o-mini": {"input": 0.00000015, "output": 0.0000006},
    "claude-3-5-sonnet": {"input": 0.000003, "output": 0.000015},
}

@dataclass
class APICall:
    model: str
    input_tokens: int
    output_tokens: int
    latency_ms: float
    cost_usd: float

def track_llm_cost(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        elapsed = (time.time() - start) * 1000

        # 사용량 추적 (실제 구현에서는 DB 저장)
        print(f"레이턴시: {elapsed:.0f}ms")
        return result
    return wrapper

import os
from langchain_core.tracers import LangChainTracer

# 환경변수 설정
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"

# 이후 모든 LangChain 호출이 자동으로 추적됨

pip install openai langchain chromadb tiktoken

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import TextLoader
import os

os.environ["OPENAI_API_KEY"] = "your-api-key"

# 책 정보 로드
loader = TextLoader("books.txt", encoding="utf-8")
documents = loader.load()

# 청킹
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
chunks = splitter.split_documents(documents)

# 벡터 DB 생성
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="books_db"
)
print(f"저장된 청크 수: {len(chunks)}")

from langchain_openai import ChatOpenAI
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory

# LLM 설정
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# 대화 메모리
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# RAG 체인
qa_chain = ConversationalRetrievalChain.from_llm(
    llm=llm,
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
    memory=memory,
    verbose=True
)

# 대화 루프
print("책 추천 챗봇입니다. 어떤 종류의 책을 찾으시나요?")
while True:
    query = input("You: ")
    if query.lower() in ["quit", "exit", "종료"]:
        break

    response = qa_chain.invoke({"question": query})
    print(f"Bot: {response['answer']}\n")