AIシステムの信頼性を高める方法｜プロンプト調整では不十分な理由と3つの設計原則

プロンプトだけでできること・できないこと

プロンプトで解決できる:
  ✅ 出力フォーマットの統一（JSON・箇条書き等）
  ✅ 語調・敬語・文体の調整
  ✅ タスクの優先順位指定
  ✅ few-shot例による精度向上（数%〜20%程度）

プロンプトでは解決できない:
  ❌ ハルシネーション（事実誤認）の根本解決
  ❌ 学習データにない最新情報への対応
  ❌ 長い処理チェーンでの状態管理
  ❌ 本番データと異なる分布への対応

設計原則1: RAGでハルシネーションを防ぐ

# reliable_rag.py
# 社内ドキュメントをベースにした信頼性の高いQ&Aシステム

import anthropic
import chromadb  # pip install chromadb
from pathlib import Path

client = anthropic.Anthropic()
chroma = chromadb.Client()
collection = chroma.create_collection("company_docs")

def add_documents(docs: list[dict]) -> None:
    """社内ドキュメントをベクトルDBに追加する"""
    collection.add(
        documents=[d["text"] for d in docs],
        metadatas=[{"source": d["source"], "date": d.get("date", "")} for d in docs],
        ids=[f"doc_{i}" for i in range(len(docs))]
    )

def ask_with_rag(question: str, n_results: int = 3) -> dict:
    """RAGを使って事実に基づいた回答を生成する"""
    # Step 1: 関連ドキュメントを検索
    results = collection.query(query_texts=[question], n_results=n_results)
    context_texts = results["documents"][0] if results["documents"] else []
    context = "\n\n".join(f"[出典: {m['source']}]\n{t}"
                          for t, m in zip(context_texts, results["metadatas"][0]))
    
    if not context:
        return {"answer": "関連する情報が見つかりませんでした。", "sources": []}
    
    # Step 2: コンテキストを使って回答を生成
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=512,
        system="""社内ドキュメントを参照して質問に答えてください。
提供されたコンテキストに情報がない場合は「情報が見つかりません」と答えてください。
推測で答えないでください。""",
        messages=[{
            "role": "user",
            "content": f"コンテキスト:\n{context}\n\n質問: {question}"
        }]
    )
    
    cost = (response.usage.input_tokens * 0.80 + response.usage.output_tokens * 4.00) / 1_000_000
    
    return {
        "answer": response.content[0].text,
        "sources": [m["source"] for m in results["metadatas"][0]],
        "cost_usd": round(cost, 5)
    }

# 使用例
add_documents([
    {"text": "Claude Sonnet 4.5の価格は入力$3/1M、出力$15/1Mです（2026年8月時点）", "source": "pricing.md"},
    {"text": "Claude Haikuは最安モデルで$0.80/$4.00/1Mです", "source": "pricing.md"},
])

result = ask_with_rag("Claude Sonnetの料金はいくらですか？")
print(f"回答: {result['answer']}")
print(f"出典: {result['sources']}")
print(f"コスト: ${result['cost_usd']:.5f}")

設計原則2: ステートマシンで処理を制御する

# state_machine_agent.py
# エージェントの状態を明示的に管理してループと暴走を防ぐ

from enum import Enum
from dataclasses import dataclass, field
from typing import Optional
import anthropic

client = anthropic.Anthropic()

class AgentState(Enum):
    IDLE = "idle"
    ANALYZING = "analyzing"
    GENERATING = "generating"
    REVIEWING = "reviewing"
    DONE = "done"
    ERROR = "error"

@dataclass
class StateMachineAgent:
    max_iterations: int = 5
    state: AgentState = AgentState.IDLE
    results: list[str] = field(default_factory=list)
    iteration: int = 0
    total_cost: float = 0.0
    
    def transition(self, new_state: AgentState) -> None:
        print(f"  {self.state.value} → {new_state.value}")
        self.state = new_state
    
    def run(self, task: str) -> str | None:
        """ステートマシンでエージェントを安全に実行する"""
        self.transition(AgentState.ANALYZING)
        
        while self.state not in [AgentState.DONE, AgentState.ERROR]:
            if self.iteration >= self.max_iterations:
                print(f"⛔ 最大反復回数({self.max_iterations})に達しました")
                self.transition(AgentState.ERROR)
                break
            
            self.iteration += 1
            print(f"Iteration {self.iteration}/{self.max_iterations}")
            
            if self.state == AgentState.ANALYZING:
                self.transition(AgentState.GENERATING)
                
            elif self.state == AgentState.GENERATING:
                response = client.messages.create(
                    model="claude-haiku-4-5-20251001",
                    max_tokens=512,
                    messages=[{"role": "user", "content": task}]
                )
                result = response.content[0].text
                cost = (response.usage.input_tokens * 0.80 + response.usage.output_tokens * 4.00) / 1_000_000
                self.total_cost += cost
                self.results.append(result)
                self.transition(AgentState.REVIEWING)
                
            elif self.state == AgentState.REVIEWING:
                if len(self.results) > 0 and len(self.results[-1]) > 50:
                    self.transition(AgentState.DONE)
                else:
                    self.transition(AgentState.GENERATING)
        
        print(f"合計コスト: ${self.total_cost:.5f}")
        return self.results[-1] if self.results else None

agent = StateMachineAgent(max_iterations=3)
result = agent.run("AI開発のベストプラクティスを3点挙げてください")
print(f"\n結果: {result[:200]}")

あわせて読みたい

• AIハルシネーション検出・対策ガイド
• Claude APIエージェント暴走対策
• AIシステム本番運用ガイド