AI DEV GUIDE SERIES

생성형 AI는 텍스트, 이미지, 코드, 음악 등 새로운 콘텐츠를 스스로 생성하는 AI입니다. LLM(Large Language Model)은 수천억 개의 파라미터를 가진 초대형 언어 모델로, 방대한 텍스트 데이터로 사전학습됩니다.

모델	개발사	파라미터	특징	API
GPT-4o	OpenAI	~1.8T	텍스트+이미지+오디오	유료/무료
Claude 3.5	Anthropic	미공개	긴 컨텍스트, 안전성	유료
Gemini 1.5	Google	미공개	1M 토큰 컨텍스트	유료/무료
Llama 3.1	Meta	8B~405B	오픈소스, 무료	자체 호스팅
EXAONE 3.5	LG AI	2.4B~32B	한국어 특화	자체 호스팅

토큰은 LLM이 텍스트를 처리하는 최소 단위입니다. 영어는 평균 4글자, 한국어는 1~2글자가 1토큰입니다.

# tiktoken으로 토큰 수 계산
    pip install tiktoken
    
    import tiktoken
    
    def count_tokens(text, model="gpt-4o"):
        enc = tiktoken.encoding_for_model(model)
        tokens = enc.encode(text)
        print(f"텍스트: {text[:50]}...")
        print(f"토큰 수: {len(tokens)}")
        print(f"예상 비용 (GPT-4o): ${len(tokens) * 0.000005:.6f}")
        return tokens
    
    count_tokens("안녕하세요! 오늘 날씨가 정말 좋네요.")
    # 토큰 수: 약 15개

모델	컨텍스트 윈도우	입력 비용	출력 비용
GPT-4o	128K 토큰	$5/1M	$15/1M
GPT-4o-mini	128K 토큰	$0.15/1M	$0.6/1M
Claude 3.5 Sonnet	200K 토큰	$3/1M	$15/1M
Gemini 1.5 Pro	1M 토큰	$3.5/1M	$10.5/1M

분야	활용 사례	주요 도구
챗봇/어시스턴트	고객 상담, 내부 Q&A 봇	OpenAI API + LangChain
문서 분석	계약서 검토, PDF 요약	RAG + Vector DB
코드 생성	자동 코드 완성, 버그 수정	Copilot API, Code Interpreter
검색 증강	사내 지식 검색, 법률 조회	RAG Pipeline
AI 에이전트	자율 작업 수행, 멀티스텝	LangGraph, AutoGen

pip install openai python-dotenv
    
    # .env 파일
    # OPENAI_API_KEY=sk-...
    
    from openai import OpenAI
    from dotenv import load_dotenv
    import os
    
    load_dotenv()
    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    
    # 기본 Chat Completion 호출
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "당신은 친절한 AI 어시스턴트입니다."},
            {"role": "user",   "content": "파이썬 리스트와 튜플의 차이를 설명해주세요."}
        ],
        temperature=0.7,    # 0: 결정론적, 1: 창의적
        max_tokens=500,     # 최대 출력 토큰
        top_p=0.9,          # 누적 확률 샘플링
        presence_penalty=0.1,  # 반복 억제
        frequency_penalty=0.1  # 빈도 기반 반복 억제
    )
    
    print(response.choices[0].message.content)
    print(f"사용 토큰: {response.usage.total_tokens}")

# 스트리밍: ChatGPT처럼 글자가 하나씩 출력
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "AI의 미래에 대해 설명해주세요."}],
        stream=True  # 스트리밍 활성화
    )
    
    full_response = ""
    for chunk in stream:
        if chunk.choices[0].delta.content is not None:
            text = chunk.choices[0].delta.content
            print(text, end="", flush=True)  # 실시간 출력
            full_response += text
    
    # FastAPI + Server-Sent Events로 웹에서 스트리밍
    from fastapi import FastAPI
    from fastapi.responses import StreamingResponse
    
    app = FastAPI()
    
    async def stream_generator(prompt: str):
        stream = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            stream=True
        )
        for chunk in stream:
            if chunk.choices[0].delta.content:
                yield f"data: {chunk.choices[0].delta.content}
    
    "
    
    @app.get("/stream")
    async def stream_response(prompt: str):
        return StreamingResponse(stream_generator(prompt),
                                  media_type="text/event-stream")

import base64
    
    # 로컬 이미지 → Base64 인코딩
    def encode_image(image_path):
        with open(image_path, "rb") as f:
            return base64.b64encode(f.read()).decode("utf-8")
    
    img_b64 = encode_image("chart.png")
    
    # Vision API 호출
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": [
                {"type": "text",  "text": "이 차트를 분석하고 주요 인사이트를 3가지 알려주세요."},
                {"type": "image_url", "image_url": {
                    "url": f"data:image/jpeg;base64,{img_b64}",
                    "detail": "high"  # low/high/auto
                }}
            ]
        }],
        max_tokens=1000
    )
    print(response.choices[0].message.content)

import numpy as np
    
    def get_embedding(text, model="text-embedding-3-small"):
        text = text.replace("
    ", " ")
        return client.embeddings.create(input=[text], model=model).data[0].embedding
    
    # 코사인 유사도 계산
    def cosine_similarity(a, b):
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
    
    # 문장 유사도 비교
    sentences = [
        "파이썬은 배우기 쉬운 프로그래밍 언어입니다.",
        "Python은 초보자에게 친숙한 언어입니다.",
        "오늘 날씨가 매우 좋습니다."
    ]
    embeddings = [get_embedding(s) for s in sentences]
    
    sim_01 = cosine_similarity(embeddings[0], embeddings[1])
    sim_02 = cosine_similarity(embeddings[0], embeddings[2])
    print(f"문장 0-1 유사도: {sim_01:.4f}")  # ~0.93 (높음)
    print(f"문장 0-2 유사도: {sim_02:.4f}")  # ~0.72 (낮음)
    
    # 배치 임베딩 (비용 절감)
    texts = ["문장1", "문장2", "문장3"]
    response = client.embeddings.create(input=texts, model="text-embedding-3-small")
    all_embeddings = [item.embedding for item in response.data]

# TTS (Text-to-Speech): 텍스트 → 음성
    response = client.audio.speech.create(
        model="tts-1-hd",
        voice="nova",     # alloy/echo/fable/onyx/nova/shimmer
        input="안녕하세요! AI 개발자 여러분을 환영합니다.",
        response_format="mp3",
        speed=1.0         # 0.25~4.0
    )
    response.stream_to_file("output.mp3")
    
    # STT (Speech-to-Text): 음성 → 텍스트 (Whisper)
    with open("audio.mp3", "rb") as audio_file:
        transcript = client.audio.transcriptions.create(
            model="whisper-1",
            file=audio_file,
            language="ko",    # 언어 지정 (없으면 자동 감지)
            response_format="verbose_json",  # 타임스탬프 포함
            timestamp_granularities=["word"]  # 단어 단위 타임스탬프
        )
    print(transcript.text)

요소	역할	예시
Role	AI의 역할 정의	"당신은 시니어 파이썬 개발자입니다"
Context	배경 정보 제공	"다음 코드는 Flask 웹 앱입니다"
Instruction	구체적 지시	"버그를 찾아 수정하고 설명해주세요"
Format	출력 형식 지정	"JSON 형식으로 반환해주세요"
Example	예시 제공	입출력 예시 1~5개

# CoT: "단계적으로 생각해" 라고 명시하면 정확도 UP
    
    def cot_prompt(question):
        return f"""다음 문제를 단계별로 생각하여 해결하세요.
    
    Q: {question}
    
    풀이 과정:
    1. 문제 파악:
    2. 핵심 정보 추출:
    3. 단계별 계산/분석:
    4. 최종 답변:"""
    
    # Zero-shot CoT: "단계별로 생각해봅시다" 추가만으로 효과
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content":
            "사과 5개가 있고 3개를 먹었습니다. 친구가 2배를 줬다면 몇 개입니까? 단계별로 생각해봅시다."
        }]
    )
    print(response.choices[0].message.content)

def few_shot_sentiment(text):
        messages = [
            {"role": "system", "content": "텍스트의 감성을 분석합니다. 긍정/부정/중립 중 하나만 출력하세요."},
            # Few-shot 예시 (학습 데이터 역할)
            {"role": "user",      "content": "이 제품 정말 최고예요!"},
            {"role": "assistant", "content": "긍정"},
            {"role": "user",      "content": "배송이 너무 느려서 실망했어요."},
            {"role": "assistant", "content": "부정"},
            {"role": "user",      "content": "평범한 제품이에요."},
            {"role": "assistant", "content": "중립"},
            # 실제 입력
            {"role": "user",      "content": text}
        ]
        response = client.chat.completions.create(model="gpt-4o-mini", messages=messages)
        return response.choices[0].message.content
    
    print(few_shot_sentiment("어제 산 노트북이 오늘 고장났어요"))  # 부정

from pydantic import BaseModel
    from typing import List
    
    # Pydantic 스키마 정의
    class ProductReview(BaseModel):
        sentiment: str          # 긍정/부정/중립
        score: float            # 1.0~5.0
        keywords: List[str]     # 핵심 키워드
        summary: str            # 한 줄 요약
    
    # Structured Output (GPT-4o 이상 지원)
    completion = client.beta.chat.completions.parse(
        model="gpt-4o-2024-08-06",
        messages=[
            {"role": "system", "content": "리뷰를 분석하여 JSON으로 반환하세요."},
            {"role": "user", "content": "배터리가 너무 빨리 닳아요. 디자인은 예쁜데 실망입니다."}
        ],
        response_format=ProductReview  # Pydantic 모델 직접 사용
    )
    
    result = completion.choices[0].message.parsed
    print(result.sentiment)   # 부정
    print(result.score)       # 2.0
    print(result.keywords)    # ["배터리", "디자인"]
    print(result.summary)     # "배터리 수명이 짧은 아쉬운 제품"

기법	효과	비용	사용 상황
Zero-shot	기본	최소	간단한 태스크
Few-shot	중간	중간	출력 형식 고정 필요
CoT	높음	중간	논리적 추론 필요
ReAct	매우 높음	높음	도구 사용 에이전트
Structured	높음 + 안정적	중간	정형 데이터 추출

컴포넌트	역할	클래스
Model	LLM/ChatModel 래퍼	ChatOpenAI, ChatAnthropic
Prompt	프롬프트 템플릿 관리	ChatPromptTemplate
Memory	대화 이력 관리	ConversationBufferMemory
Chain	컴포넌트 연결	LLMChain, LCEL (|)
Tools	외부 도구 연동	Tool, @tool 데코레이터

pip install langchain langchain-openai langchain-community
    
    from langchain_openai import ChatOpenAI
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.output_parsers import StrOutputParser
    from langchain_core.runnables import RunnableParallel, RunnablePassthrough
    
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
    
    # === LCEL: 파이프(|)로 체인 구성 ===
    prompt = ChatPromptTemplate.from_messages([
        ("system", "당신은 {topic} 전문가입니다."),
        ("human",  "{question}")
    ])
    
    chain = prompt | llm | StrOutputParser()
    
    result = chain.invoke({"topic": "파이썬", "question": "제너레이터를 설명해주세요"})
    print(result)
    
    # 병렬 처리
    parallel_chain = RunnableParallel({
        "korean": ChatPromptTemplate.from_template("다음을 한국어로: {text}") | llm | StrOutputParser(),
        "english": ChatPromptTemplate.from_template("Translate to English: {text}") | llm | StrOutputParser()
    })
    result = parallel_chain.invoke({"text": "AI는 미래입니다"})
    print(result["korean"])
    print(result["english"])

from langchain_community.chat_message_histories import ChatMessageHistory
    from langchain_core.runnables.history import RunnableWithMessageHistory
    
    # 대화 이력 저장소
    store = {}  # 실제 서비스는 Redis/DB 사용
    
    def get_session_history(session_id: str) -> ChatMessageHistory:
        if session_id not in store:
            store[session_id] = ChatMessageHistory()
        return store[session_id]
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", "당신은 친절한 AI 상담사입니다."),
        ("placeholder", "{chat_history}"),  # 이력 삽입
        ("human", "{input}")
    ])
    
    chain = prompt | llm | StrOutputParser()
    
    with_history = RunnableWithMessageHistory(
        chain,
        get_session_history,
        input_messages_key="input",
        history_messages_key="chat_history"
    )
    
    # 세션별 대화
    config = {"configurable": {"session_id": "user123"}}
    print(with_history.invoke({"input": "안녕하세요! 파이썬 배우고 싶어요"}, config=config))
    print(with_history.invoke({"input": "어디서부터 시작하면 좋을까요?"}, config=config))
    # 두 번째 호출 시 첫 대화를 기억함!

from langchain.tools import tool
    from langchain.agents import create_openai_tools_agent, AgentExecutor
    
    # 커스텀 도구 정의
    @tool
    def get_weather(city: str) -> str:
        """도시의 현재 날씨를 가져옵니다."""
        # 실제 API 연동 부분
        return f"{city}의 현재 날씨: 맑음, 23도"
    
    @tool
    def calculate(expression: str) -> str:
        """수학 식을 계산합니다. 예: 2+3*4"""
        try:
            return str(eval(expression))
        except Exception as e:
            return f"계산 오류: {e}"
    
    @tool
    def search_web(query: str) -> str:
        """인터넷에서 정보를 검색합니다."""
        # 실제로는 Tavily, SerpAPI 등 연동
        return f"검색 결과: {query}에 대한 정보..."
    
    tools = [get_weather, calculate, search_web]
    
    # 에이전트 생성
    prompt = ChatPromptTemplate.from_messages([
        ("system", "당신은 도구를 사용할 수 있는 AI 어시스턴트입니다."),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}")
    ])
    agent = create_openai_tools_agent(llm, tools, prompt)
    executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
    
    result = executor.invoke({"input": "서울 날씨 알려주고, 23+47 계산해줘"})
    print(result["output"])

DB	타입	스케일	비용	특징
Chroma	로컬	소~중	무료	개발/프로토타입 최적
FAISS	로컬	대규모	무료	Meta 개발, GPU 가속
Pinecone	클라우드	무제한	유료	관리형, 프로덕션
Weaviate	클라우드/로컬	대규모	무료/유료	GraphQL, 하이브리드 검색
Qdrant	클라우드/로컬	대규모	무료/유료	Rust 기반, 고성능

pip install chromadb langchain-chroma
    
    import chromadb
    from langchain_chroma import Chroma
    from langchain_openai import OpenAIEmbeddings
    from langchain.text_splitter import RecursiveCharacterTextSplitter
    
    # 임베딩 모델
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    
    # 텍스트 청킹
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=500,       # 청크 크기 (문자 수)
        chunk_overlap=50,     # 겹치는 부분 (문맥 유지)
        separators=["
    
    ", "
    ", " ", ""]  # 분리 기준
    )
    
    # 문서 로드 및 분할
    from langchain_community.document_loaders import PyPDFLoader
    loader = PyPDFLoader("manual.pdf")
    documents = loader.load()
    chunks = splitter.split_documents(documents)
    print(f"총 {len(chunks)}개 청크 생성")
    
    # Vector DB 생성 및 저장
    vectordb = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory="./chroma_db"  # 영구 저장
    )
    
    # 유사도 검색
    query = "환불 정책이 어떻게 되나요?"
    results = vectordb.similarity_search_with_score(query, k=3)
    for doc, score in results:
        print(f"유사도: {score:.4f}")
        print(f"내용: {doc.page_content[:100]}")
        print("---")

pip install faiss-cpu  # CPU 버전
    # pip install faiss-gpu  # GPU 버전
    
    from langchain_community.vectorstores import FAISS
    
    # FAISS 생성
    faiss_db = FAISS.from_documents(chunks, embeddings)
    
    # 로컬 저장/로드
    faiss_db.save_local("./faiss_index")
    loaded_db = FAISS.load_local("./faiss_index", embeddings,
                                   allow_dangerous_deserialization=True)
    
    # MMR 검색 (Maximum Marginal Relevance) — 다양성 보장
    results = faiss_db.max_marginal_relevance_search(
        query="파이썬 기초 학습 방법",
        k=4,          # 반환 개수
        fetch_k=20,   # 후보 풀
        lambda_mult=0.5  # 0: 다양성, 1: 유사도 우선
    )
    
    # 메타데이터 필터링
    results = faiss_db.similarity_search(
        query, k=3,
        filter={"source": "manual.pdf", "page": {"$gte": 10}}
    )

RAG(Retrieval-Augmented Generation)는 외부 지식을 검색하여 LLM에 제공하는 방식입니다. LLM의 학습 데이터 한계와 환각 문제를 해결합니다.

단계	작업	도구
1. Indexing	문서 로드 → 청킹 → 임베딩 → 저장	DocumentLoader + TextSplitter
2. Retrieval	쿼리 임베딩 → 유사도 검색	VectorStore.as_retriever()
3. Generation	컨텍스트 + 쿼리 → LLM → 답변	RetrievalQA Chain

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
    from langchain_chroma import Chroma
    from langchain_core.prompts import ChatPromptTemplate
    from langchain_core.runnables import RunnablePassthrough
    from langchain_core.output_parsers import StrOutputParser
    from langchain.text_splitter import RecursiveCharacterTextSplitter
    from langchain_community.document_loaders import PyPDFLoader
    
    # === STEP 1: 인덱싱 ===
    loader = PyPDFLoader("company_manual.pdf")
    docs = loader.load()
    splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
    chunks = splitter.split_documents(docs)
    
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectordb = Chroma.from_documents(chunks, embeddings, persist_directory="./db")
    
    # === STEP 2: 리트리버 ===
    retriever = vectordb.as_retriever(
        search_type="similarity",
        search_kwargs={"k": 4}  # 상위 4개 청크 가져오기
    )
    
    # === STEP 3: 생성 체인 ===
    RAG_PROMPT = ChatPromptTemplate.from_template("""
    당신은 주어진 컨텍스트만을 기반으로 질문에 답변하는 AI입니다.
    컨텍스트에 없는 정보는 "모르겠습니다"라고 답하세요.
    
    컨텍스트:
    {context}
    
    질문: {question}
    
    답변:""")
    
    def format_docs(docs):
        return "
    
    ".join(doc.page_content for doc in docs)
    
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    
    rag_chain = (
        {"context": retriever | format_docs, "question": RunnablePassthrough()}
        | RAG_PROMPT
        | llm
        | StrOutputParser()
    )
    
    answer = rag_chain.invoke("연차 신청 방법이 어떻게 되나요?")
    print(answer)

# === HyDE (Hypothetical Document Embeddings) ===
    # 실제 문서 검색 전에 가상 답변을 먼저 생성하여 검색 정확도 향상
    
    from langchain.retrievers import HypotheticalDocumentEmbedder
    
    hyde_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.5)
    hyde_prompt = ChatPromptTemplate.from_template(
        "다음 질문에 대한 모범 답변 예시를 작성하세요: {question}"
    )
    hyde_chain = hyde_prompt | hyde_llm | StrOutputParser()
    
    def hyde_retriever(question):
        hypothetical_doc = hyde_chain.invoke({"question": question})
        return retriever.invoke(hypothetical_doc)  # 가상 답변으로 검색
    
    # === Re-ranking (Cohere) ===
    # 검색된 청크를 재순위화하여 가장 관련성 높은 것 선별
    pip install cohere
    
    from langchain.retrievers.contextual_compression import ContextualCompressionRetriever
    from langchain_cohere import CohereRerank
    
    compressor = CohereRerank(model="rerank-multilingual-v3.0", top_n=3)
    compression_retriever = ContextualCompressionRetriever(
        base_compressor=compressor,
        base_retriever=vectordb.as_retriever(search_kwargs={"k": 10})
    )
    
    # === 앙상블 리트리버 ===
    # 키워드 검색 + 벡터 검색을 함께 사용 (Hybrid Search)
    from langchain_community.retrievers import BM25Retriever
    from langchain.retrievers import EnsembleRetriever
    
    bm25 = BM25Retriever.from_documents(chunks, k=4)  # 키워드 기반
    vector = vectordb.as_retriever(search_kwargs={"k": 4})  # 의미 기반
    
    ensemble = EnsembleRetriever(
        retrievers=[bm25, vector],
        weights=[0.4, 0.6]  # 키워드 40%, 벡터 60%
    )

지표	설명	도구
Faithfulness	답변이 컨텍스트에 충실한가	RAGAS
Answer Relevance	답변이 질문에 관련 있는가	RAGAS
Context Recall	필요한 정보를 모두 검색했는가	RAGAS
Context Precision	검색된 정보가 관련 있는가	RAGAS

import json
    from openai import OpenAI
    
    client = OpenAI()
    
    # 도구 스키마 정의 (JSON Schema)
    tools = [
        {
            "type": "function",
            "function": {
                "name": "get_stock_price",
                "description": "주식 종목의 현재 가격을 가져옵니다",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "symbol": {"type": "string", "description": "주식 티커 (예: AAPL)"},
                        "currency": {"type": "string", "enum": ["USD", "KRW"], "description": "통화"}
                    },
                    "required": ["symbol"]
                }
            }
        },
        {
            "type": "function",
            "function": {
                "name": "send_email",
                "description": "이메일을 발송합니다",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "to": {"type": "string"},
                        "subject": {"type": "string"},
                        "body": {"type": "string"}
                    },
                    "required": ["to", "subject", "body"]
                }
            }
        }
    ]
    
    # 실제 함수 구현
    def get_stock_price(symbol, currency="USD"):
        # 실제로는 Yahoo Finance API 등 호출
        prices = {"AAPL": 195.0, "TSLA": 250.0, "NVDA": 875.0}
        price = prices.get(symbol, 0)
        if currency == "KRW": price *= 1350
        return {"symbol": symbol, "price": price, "currency": currency}
    
    def send_email(to, subject, body):
        print(f"이메일 발송: {to} | 제목: {subject}")
        return {"status": "sent", "message_id": "msg_123"}
    
    # 함수 디스패처
    function_map = {"get_stock_price": get_stock_price, "send_email": send_email}
    
    # 멀티턴 Function Calling 루프
    def run_agent(user_message):
        messages = [{"role": "user", "content": user_message}]
    
        while True:
            response = client.chat.completions.create(
                model="gpt-4o",
                messages=messages,
                tools=tools,
                tool_choice="auto"
            )
            msg = response.choices[0].message
            messages.append(msg)
    
            if response.choices[0].finish_reason == "tool_calls":
                for call in msg.tool_calls:
                    args = json.loads(call.function.arguments)
                    result = function_map[call.function.name](**args)
                    messages.append({"role": "tool", "tool_call_id": call.id,
                                      "content": json.dumps(result, ensure_ascii=False)})
            else:
                return msg.content  # 최종 답변
    
    print(run_agent("애플 주가 알려주고 CEO에게 메일 보내줘"))

pip install langgraph
    
    from langgraph.graph import StateGraph, END
    from typing import TypedDict, List
    
    # 상태 정의
    class AgentState(TypedDict):
        messages: List[dict]
        next_step: str
        final_answer: str
    
    # 노드 함수들
    def analyze_node(state: AgentState) -> AgentState:
        """사용자 의도 분석"""
        user_msg = state["messages"][-1]["content"]
        # LLM으로 의도 분류
        if "주가" in user_msg or "stock" in user_msg:
            state["next_step"] = "stock_agent"
        elif "날씨" in user_msg:
            state["next_step"] = "weather_agent"
        else:
            state["next_step"] = "general_agent"
        return state
    
    def stock_agent_node(state: AgentState) -> AgentState:
        """주식 정보 처리 에이전트"""
        result = run_agent(state["messages"][-1]["content"])
        state["final_answer"] = result
        state["next_step"] = "end"
        return state
    
    # 그래프 구성
    workflow = StateGraph(AgentState)
    workflow.add_node("analyze", analyze_node)
    workflow.add_node("stock_agent", stock_agent_node)
    workflow.set_entry_point("analyze")
    
    # 조건부 엣지
    workflow.add_conditional_edges(
        "analyze",
        lambda s: s["next_step"],
        {"stock_agent": "stock_agent", "end": END}
    )
    workflow.add_edge("stock_agent", END)
    
    app = workflow.compile()
    result = app.invoke({"messages": [{"role": "user", "content": "삼성전자 주가 알려줘"}], "next_step": "", "final_answer": ""})
    print(result["final_answer"])

항목	Fine-tuning	RAG
정보 유형	스타일, 형식, 행동 패턴	최신 정보, 사실적 지식
데이터 필요량	수십~수천 예시	문서 바로 사용
비용	학습 비용 발생	API 비용만
실시간 업데이트	재학습 필요	문서 추가만
추천 상황	특정 말투, 전문 분야	지식 기반 Q&A

# 1. 학습 데이터 준비 (JSONL 형식)
    import json
    
    training_data = [
        {
            "messages": [
                {"role": "system", "content": "당신은 한국 음식 전문 셰프입니다."},
                {"role": "user", "content": "김치찌개 레시피 알려주세요"},
                {"role": "assistant", "content": "재료: 김치 300g, 돼지고기 200g..."}
            ]
        },
        # ... 최소 10개 이상, 권장 100개+
    ]
    
    with open("training.jsonl", "w", encoding="utf-8") as f:
        for item in training_data:
            f.write(json.dumps(item, ensure_ascii=False) + "
    ")
    
    # 2. 파일 업로드
    with open("training.jsonl", "rb") as f:
        file_obj = client.files.create(file=f, purpose="fine-tune")
    print(f"파일 ID: {file_obj.id}")
    
    # 3. 파인튜닝 시작
    ft_job = client.fine_tuning.jobs.create(
        training_file=file_obj.id,
        model="gpt-4o-mini-2024-07-18",
        hyperparameters={
            "n_epochs": 3,
            "batch_size": 4,
            "learning_rate_multiplier": 2
        }
    )
    print(f"Job ID: {ft_job.id}")
    
    # 4. 상태 확인
    job_status = client.fine_tuning.jobs.retrieve(ft_job.id)
    print(f"상태: {job_status.status}")  # running/succeeded/failed
    
    # 5. 파인튜닝된 모델 사용
    response = client.chat.completions.create(
        model=job_status.fine_tuned_model,  # "ft:gpt-4o-mini:..."
        messages=[{"role": "user", "content": "불고기 레시피"}]
    )
    print(response.choices[0].message.content)

# Unsloth: 2배 빠른 LoRA 파인튜닝 (Google Colab 추천)
    pip install unsloth
    
    from unsloth import FastLanguageModel
    import torch
    
    # 모델 로드 (4비트 양자화)
    model, tokenizer = FastLanguageModel.from_pretrained(
        model_name="unsloth/llama-3-8b-Instruct-bnb-4bit",
        max_seq_length=2048,
        load_in_4bit=True  # 메모리 절약
    )
    
    # LoRA 설정
    model = FastLanguageModel.get_peft_model(
        model,
        r=16,                              # LoRA 랭크
        target_modules=["q_proj", "k_proj", "v_proj", "o_proj",
                         "gate_proj", "up_proj", "down_proj"],
        lora_alpha=16,
        lora_dropout=0,
        bias="none",
        use_gradient_checkpointing="unsloth"
    )
    
    # Alpaca 형식 데이터셋 학습
    from trl import SFTTrainer
    from transformers import TrainingArguments
    from datasets import load_dataset
    
    dataset = load_dataset("yahma/alpaca-cleaned", split="train")
    
    trainer = SFTTrainer(
        model=model,
        tokenizer=tokenizer,
        train_dataset=dataset,
        dataset_text_field="text",
        max_seq_length=2048,
        args=TrainingArguments(
            per_device_train_batch_size=2,
            gradient_accumulation_steps=4,
            num_train_epochs=1,
            learning_rate=2e-4,
            output_dir="outputs"
        )
    )
    trainer.train()
    
    # GGUF 형식으로 저장 (Ollama 배포용)
    model.save_pretrained_gguf("model_gguf", tokenizer, quantization_method="q4_k_m")

# Streamlit + RAG + LangChain 사내 챗봇
    pip install streamlit langchain-openai langchain-chroma pypdf
    
    import streamlit as st
    from langchain_openai import ChatOpenAI, OpenAIEmbeddings
    from langchain_chroma import Chroma
    from langchain_community.document_loaders import PyPDFDirectoryLoader
    from langchain.text_splitter import RecursiveCharacterTextSplitter
    from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
    from langchain_core.runnables.history import RunnableWithMessageHistory
    from langchain_community.chat_message_histories import StreamlitChatMessageHistory
    from langchain_core.output_parsers import StrOutputParser
    from langchain_core.runnables import RunnablePassthrough
    
    st.title("사내 문서 Q&A 챗봇")
    
    # 사이드바: PDF 업로드
    uploaded_files = st.sidebar.file_uploader("PDF 업로드", type="pdf", accept_multiple_files=True)
    
    @st.cache_resource
    def build_vectordb(_files):
        docs = []
        for f in _files:
            from langchain_community.document_loaders import PyPDFLoader
            import tempfile, os
            with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as tmp:
                tmp.write(f.read())
                docs.extend(PyPDFLoader(tmp.name).load())
        splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
        chunks = splitter.split_documents(docs)
        return Chroma.from_documents(chunks, OpenAIEmbeddings(model="text-embedding-3-small"))
    
    if uploaded_files:
        vectordb = build_vectordb(tuple(uploaded_files))
        retriever = vectordb.as_retriever(search_kwargs={"k": 4})
    
        prompt = ChatPromptTemplate.from_messages([
            ("system", "컨텍스트를 기반으로 질문에 답하세요. 없으면 모른다고 하세요.
    
    컨텍스트: {context}"),
            MessagesPlaceholder("history"),
            ("human", "{question}")
        ])
    
        llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, streaming=True)
        history = StreamlitChatMessageHistory()
    
        chain = (
            {"context": retriever | (lambda docs: "
    ".join(d.page_content for d in docs)),
             "question": RunnablePassthrough(), "history": lambda _: history.messages}
            | prompt | llm | StrOutputParser()
        )
    
        # 채팅 UI
        for msg in history.messages:
            st.chat_message(msg.type).write(msg.content)
    
        if question := st.chat_input("질문을 입력하세요"):
            st.chat_message("user").write(question)
            with st.chat_message("assistant"):
                response = st.write_stream(chain.stream(question))
            history.add_user_message(question)
            history.add_ai_message(response)

# FastAPI로 GitHub Webhook 처리
    from fastapi import FastAPI, Request
    from openai import OpenAI
    import httpx
    
    app = FastAPI()
    client = OpenAI()
    
    CODE_REVIEW_PROMPT = """당신은 시니어 개발자입니다. 다음 코드를 리뷰하고 아래 항목을 평가하세요:
    1. 코드 품질 (0-10점)
    2. 잠재적 버그
    3. 성능 개선 포인트
    4. 보안 취약점
    5. 개선 코드 제안
    
    코드:
    {code}"""
    
    @app.post("/webhook/github")
    async def handle_pr(request: Request):
        payload = await request.json()
        if payload.get("action") not in ["opened", "synchronize"]:
            return {"status": "ignored"}
    
        pr = payload["pull_request"]
        diff_url = pr["diff_url"]
    
        # PR diff 가져오기
        async with httpx.AsyncClient() as h:
            resp = await h.get(diff_url, headers={"Authorization": f"token {GITHUB_TOKEN}"})
            diff = resp.text[:4000]  # 토큰 제한
    
        # AI 리뷰 생성
        review = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": CODE_REVIEW_PROMPT.format(code=diff)}]
        ).choices[0].message.content
    
        # PR에 코멘트 달기
        comment_url = pr["comments_url"]
        async with httpx.AsyncClient() as h:
            await h.post(comment_url,
                json={"body": f"## AI 코드 리뷰
    
    {review}"},
                headers={"Authorization": f"token {GITHUB_TOKEN}",
                          "Accept": "application/vnd.github.v3+json"}
            )
        return {"status": "reviewed"}

전략	효과	구현 방법
모델 캐싱	동일 쿼리 API 재호출 방지	Redis + 쿼리 해시
모델 라우팅	복잡도별 모델 선택	간단→mini, 복잡→4o
프롬프트 압축	토큰 수 감소	LLMLingua
배치 처리	API 호출 횟수 감소	Batch API (50% 할인)
로컬 LLM	API 비용 0	Ollama + Llama 3

이 가이드를 완료했다면 생성형 AI 서비스를 기획하고 직접 구현·배포할 수 있는 실력이 됩니다.
마지막 가이드에서는 HuggingFace, FastAPI, Docker, CI/CD로 실제 프로덕션 배포를 다룹니다.