AI 스타트업 & 제품 개발 가이드: LLM API부터 스케일링, 비즈니스 모델까지

# Ollama 설치 후 모델 실행
ollama pull llama3.3
ollama run llama3.3

pip install vllm

python -m vllm.entrypoints.openai.api_server \
  --model meta-llama/Llama-3.3-70B-Instruct \
  --tensor-parallel-size 4 \
  --gpu-memory-utilization 0.9

예산이 제한적인가?
├── Yes → 오픈소스 (Llama, Mistral) 자체 호스팅
│         또는 Gemini Flash (저비용 API)
└── No  → 품질 요구사항 높은가?
          ├── Yes → Claude 3.5 Sonnet / GPT-4o
          └── No  → GPT-4o mini / Claude Haiku

import streamlit as st
from openai import OpenAI

client = OpenAI()

st.title("AI 문서 요약기 MVP")

uploaded_file = st.file_uploader("문서를 업로드하세요", type=["txt", "pdf"])
tone = st.selectbox("요약 톤", ["비즈니스", "캐주얼", "기술적"])

if uploaded_file and st.button("요약 시작"):
    text = uploaded_file.read().decode("utf-8")

    with st.spinner("AI가 요약 중..."):
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {
                    "role": "system",
                    "content": f"당신은 전문 문서 요약가입니다. {tone} 톤으로 요약해주세요."
                },
                {
                    "role": "user",
                    "content": f"다음 문서를 3-5문장으로 요약해주세요:\n\n{text[:4000]}"
                }
            ],
            max_tokens=500
        )

    summary = response.choices[0].message.content
    st.success("요약 완료!")
    st.write(summary)

    st.download_button(
        label="요약 다운로드",
        data=summary,
        file_name="summary.txt"
    )

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA

# 문서 분할
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
chunks = splitter.split_documents(documents)

# 벡터 스토어 생성
vectorstore = Chroma.from_documents(
    chunks,
    OpenAIEmbeddings()
)

# RAG 체인 구성
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),
    return_source_documents=True
)

result = qa_chain.invoke({"query": "제품의 환불 정책은?"})
print(result["result"])

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

judge_llm = ChatOpenAI(model="gpt-4o", temperature=0)

EVAL_PROMPT = ChatPromptTemplate.from_template("""
당신은 AI 응답 품질 평가자입니다.

질문: {question}
AI 응답: {response}
기준 답변: {reference}

다음 기준으로 1-5점 평가하세요:
- 정확성 (Accuracy): 사실적으로 정확한가?
- 완전성 (Completeness): 질문에 충분히 답했는가?
- 명확성 (Clarity): 이해하기 쉬운가?

JSON 형식으로 응답: {{"accuracy": 점수, "completeness": 점수, "clarity": 점수, "reasoning": "설명"}}
""")

def evaluate_response(question, response, reference):
    chain = EVAL_PROMPT | judge_llm
    result = chain.invoke({
        "question": question,
        "response": response,
        "reference": reference
    })
    return result.content

import random
from dataclasses import dataclass
from typing import Dict, List
import json

@dataclass
class PromptVariant:
    name: str
    system_prompt: str
    wins: int = 0
    total: int = 0

    @property
    def win_rate(self):
        return self.wins / self.total if self.total > 0 else 0

class PromptABTest:
    def __init__(self, variants: List[PromptVariant]):
        self.variants = {v.name: v for v in variants}

    def select_variant(self) -> PromptVariant:
        # 엡실론-그리디: 10% 탐색, 90% 활용
        if random.random() < 0.1:
            return random.choice(list(self.variants.values()))
        return max(self.variants.values(), key=lambda v: v.win_rate)

    def record_feedback(self, variant_name: str, positive: bool):
        v = self.variants[variant_name]
        v.total += 1
        if positive:
            v.wins += 1

    def report(self) -> Dict:
        return {
            name: {"win_rate": f"{v.win_rate:.1%}", "total": v.total}
            for name, v in self.variants.items()
        }

# 사용 예시
ab_test = PromptABTest([
    PromptVariant("formal", "당신은 전문적이고 격식 있는 AI 어시스턴트입니다."),
    PromptVariant("casual", "안녕하세요! 친근하고 쉽게 설명하는 AI입니다."),
])

from fastapi import FastAPI
from pydantic import BaseModel
from datetime import datetime
import json

app = FastAPI()

class FeedbackRequest(BaseModel):
    session_id: str
    message_id: str
    rating: int  # 1-5
    comment: str = ""
    prompt_variant: str = "default"

feedback_store = []

@app.post("/feedback")
async def collect_feedback(feedback: FeedbackRequest):
    entry = {
        "timestamp": datetime.utcnow().isoformat(),
        **feedback.dict()
    }
    feedback_store.append(entry)

    # 실제 환경에서는 DB 저장
    with open("feedback_log.jsonl", "a") as f:
        f.write(json.dumps(entry) + "\n")

    return {"status": "recorded", "message_id": feedback.message_id}

@app.get("/feedback/stats")
async def get_stats():
    if not feedback_store:
        return {"avg_rating": 0, "total": 0}

    avg = sum(f["rating"] for f in feedback_store) / len(feedback_store)
    return {
        "avg_rating": round(avg, 2),
        "total": len(feedback_store),
        "positive_rate": f"{sum(1 for f in feedback_store if f['rating'] >= 4) / len(feedback_store):.1%}"
    }

from dataclasses import dataclass
from typing import Dict

@dataclass
class ModelPricing:
    input_per_1m: float   # USD per 1M input tokens
    output_per_1m: float  # USD per 1M output tokens
    cache_write_per_1m: float = 0.0
    cache_read_per_1m: float = 0.0

PRICING: Dict[str, ModelPricing] = {
    "gpt-4o": ModelPricing(2.50, 10.00),
    "gpt-4o-mini": ModelPricing(0.15, 0.60),
    "claude-3-5-sonnet": ModelPricing(3.00, 15.00, 3.75, 0.30),
    "claude-3-haiku": ModelPricing(0.25, 1.25, 0.30, 0.03),
    "gemini-2.0-flash": ModelPricing(0.075, 0.30),
}

def calculate_monthly_cost(
    model: str,
    daily_requests: int,
    avg_input_tokens: int,
    avg_output_tokens: int,
    cache_hit_rate: float = 0.0
) -> dict:
    p = PRICING[model]
    monthly_requests = daily_requests * 30

    # 캐시 히트를 고려한 실제 입력 토큰
    cached_tokens = avg_input_tokens * cache_hit_rate
    fresh_tokens = avg_input_tokens * (1 - cache_hit_rate)

    input_cost = (fresh_tokens * monthly_requests / 1_000_000) * p.input_per_1m
    cache_read_cost = (cached_tokens * monthly_requests / 1_000_000) * p.cache_read_per_1m
    output_cost = (avg_output_tokens * monthly_requests / 1_000_000) * p.output_per_1m

    total = input_cost + cache_read_cost + output_cost

    return {
        "model": model,
        "monthly_requests": monthly_requests,
        "total_usd": round(total, 2),
        "cost_per_request_usd": round(total / monthly_requests, 6),
        "breakdown": {
            "input": round(input_cost, 2),
            "cache_read": round(cache_read_cost, 2),
            "output": round(output_cost, 2)
        }
    }

# 예시: 하루 10,000 요청
result = calculate_monthly_cost(
    model="claude-3-5-sonnet",
    daily_requests=10_000,
    avg_input_tokens=2000,
    avg_output_tokens=500,
    cache_hit_rate=0.7  # 70% 캐시 히트율
)
print(f"월 예상 비용: ${result['total_usd']}")

import anthropic

client = anthropic.Anthropic()

# 시스템 프롬프트를 캐시에 저장
response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "당신은 전문 법률 AI 어시스턴트입니다.",
        },
        {
            "type": "text",
            "text": open("legal_knowledge_base.txt").read(),  # 대용량 컨텍스트
            "cache_control": {"type": "ephemeral"}  # 캐시 마킹
        }
    ],
    messages=[
        {"role": "user", "content": "계약서 해지 조건을 설명해주세요."}
    ]
)

# cache_read_input_tokens로 캐시 효율 확인
usage = response.usage
print(f"캐시 읽기 토큰: {usage.cache_read_input_tokens}")
print(f"캐시 쓰기 토큰: {usage.cache_creation_input_tokens}")

from openai import OpenAI

client = OpenAI()

def grounded_response(query: str, context: str) -> dict:
    """RAG 기반 근거 있는 응답 생성"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": """반드시 제공된 컨텍스트에만 근거하여 답하세요.
                컨텍스트에 없는 정보는 '제공된 정보에 없습니다'라고 명확히 말하세요.
                확실하지 않으면 '확인이 필요합니다'라고 답하세요."""
            },
            {
                "role": "user",
                "content": f"컨텍스트:\n{context}\n\n질문: {query}"
            }
        ],
        temperature=0,  # 결정론적 응답
        logprobs=True   # 신뢰도 확인용
    )

    message = response.choices[0]
    # logprob으로 불확실성 추정
    avg_logprob = sum(
        t.logprob for t in message.logprobs.content
    ) / len(message.logprobs.content) if message.logprobs else 0

    return {
        "answer": message.message.content,
        "confidence": round(2.718 ** avg_logprob, 3),  # e^logprob = prob
        "needs_review": avg_logprob < -1.5  # 낮은 신뢰도 플래그
    }