ai-engineer

AI エンジニアリング

本番環境向けの LLM アプリケーションと AI システムを構築します。

使用する場面

• LLM API の統合
• RAG システムの構築
• AI エージェントの作成
• ベクトルデータベースのセットアップ
• トークンの最適化

LLM の統合

API のセットアップ

from anthropic import Anthropic

client = Anthropic()

def chat(messages: list[dict], system: str = None) -> str:
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        system=system or "You are a helpful assistant.",
        messages=messages
    )
    return response.content[0].text

# With retry and error handling
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10))
def safe_chat(messages, system=None):
    try:
        return chat(messages, system)
    except Exception as e:
        logger.error(f"LLM call failed: {e}")
        raise

構造化出力

import json

def extract_structured(text: str, schema: dict) -> dict:
    prompt = f"""Extract information from the text according to this schema:
{json.dumps(schema, indent=2)}

Text: {text}

Return valid JSON only."""

    response = chat([{"role": "user", "content": prompt}])
    return json.loads(response)

RAG システム

ドキュメント処理

from langchain.text_splitter import RecursiveCharacterTextSplitter

def chunk_documents(docs: list[str], chunk_size=1000, overlap=200):
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        separators=["\n\n", "\n", ". ", " "]
    )
    return splitter.split_documents(docs)

ベクトルストア

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams

client = QdrantClient(":memory:")  # or url="http://localhost:6333"

# Create collection
client.create_collection(
    collection_name="docs",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE)
)

# Upsert vectors
client.upsert(
    collection_name="docs",
    points=[
        {"id": i, "vector": embed(chunk), "payload": {"text": chunk}}
        for i, chunk in enumerate(chunks)
    ]
)

# Search
results = client.search(
    collection_name="docs",
    query_vector=embed(query),
    limit=5
)

RAG クエリ

def rag_query(question: str, top_k=5) -> str:
    # Retrieve relevant chunks
    results = client.search(
        collection_name="docs",
        query_vector=embed(question),
        limit=top_k
    )

    context = "\n\n".join([r.payload["text"] for r in results])

    prompt = f"""Answer based on the context below.

Context:
{context}

Question: {question}

Answer:"""

    return chat([{"role": "user", "content": prompt}])

コスト最適化

• 頻繁に使うクエリをキャッシュします
• 簡単なタスクには小型のモデルを使用します
• 可能な場合はリクエストをバッチ処理します
• 機能ごとにトークンの使用量を追跡します
• max_tokens を適切に設定します

例

入力：「このアプリに AI チャットを追加してください」 **アクション：**LLM クライアントをセットアップし、チャットエンドポイントを作成して、エラー処理を追加します

入力：「ドキュメント向けの RAG を構築してください」 **アクション：**ドキュメントを分割し、埋め込みを作成して、ベクトルストアをセットアップし、検索を実装します