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"],  # Attentionレイヤーにのみ適用
    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")