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

# After installing Ollama, pull and run a model
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

Is budget constrained?
├── Yes → Open source (Llama, Mistral) self-hosted
│         OR Gemini Flash (low-cost API)
└── No  → Are quality requirements high?
          ├── 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 Document Summarizer MVP")

uploaded_file = st.file_uploader("Upload a document", type=["txt", "pdf"])
tone = st.selectbox("Summary tone", ["Business", "Casual", "Technical"])

if uploaded_file and st.button("Summarize"):
    text = uploaded_file.read().decode("utf-8")

    with st.spinner("AI is summarizing..."):
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {
                    "role": "system",
                    "content": f"You are a professional document summarizer. Use a {tone} tone."
                },
                {
                    "role": "user",
                    "content": f"Summarize the following document in 3-5 sentences:\n\n{text[:4000]}"
                }
            ],
            max_tokens=500
        )

    summary = response.choices[0].message.content
    st.success("Summary complete!")
    st.write(summary)

    st.download_button(
        label="Download summary",
        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

# Split documents
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
chunks = splitter.split_documents(documents)

# Create vector store
vectorstore = Chroma.from_documents(
    chunks,
    OpenAIEmbeddings()
)

# Build RAG chain
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": "What is the refund policy?"})
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("""
You are an AI response quality evaluator.

Question: {question}
AI Response: {response}
Reference Answer: {reference}

Score each criterion from 1 to 5:
- Accuracy: Is the response factually correct?
- Completeness: Does it fully address the question?
- Clarity: Is it easy to understand?

Respond in JSON: {{"accuracy": score, "completeness": score, "clarity": score, "reasoning": "explanation"}}
""")

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

@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:
        # Epsilon-greedy: 10% exploration, 90% exploitation
        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()
        }

# Usage
ab_test = PromptABTest([
    PromptVariant("formal", "You are a professional and formal AI assistant."),
    PromptVariant("casual", "Hey! I'm a friendly AI that explains things simply."),
])

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)

    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)
        }
    }

# Example: 10,000 requests per day
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% cache hit rate
)
print(f"Estimated monthly cost: ${result['total_usd']}")

import anthropic

client = anthropic.Anthropic()

# Cache the system prompt and large context
response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "You are a professional legal AI assistant.",
        },
        {
            "type": "text",
            "text": open("legal_knowledge_base.txt").read(),  # large context
            "cache_control": {"type": "ephemeral"}  # mark for caching
        }
    ],
    messages=[
        {"role": "user", "content": "Explain the conditions for contract termination."}
    ]
)

usage = response.usage
print(f"Cache read tokens: {usage.cache_read_input_tokens}")
print(f"Cache write tokens: {usage.cache_creation_input_tokens}")

from openai import OpenAI

client = OpenAI()

def grounded_response(query: str, context: str) -> dict:
    """Generate a grounded response using RAG context."""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": """Answer only based on the provided context.
                If information is not in the context, say 'Not found in provided information.'
                If you are uncertain, say 'Verification required.'"""
            },
            {
                "role": "user",
                "content": f"Context:\n{context}\n\nQuestion: {query}"
            }
        ],
        temperature=0,   # deterministic
        logprobs=True    # for confidence scoring
    )

    message = response.choices[0]
    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),
        "needs_review": avg_logprob < -1.5
    }