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

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

    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   # 1Mトークンあたりの価格（USD）
    output_per_1m: float
    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)
        }
    }

# 例：1日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": "契約解除の条件を説明してください。"}
    ]
)

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