【Lost in the Middle の精度分布 (U字型曲線)】

 100% |  ██ (高精度)                                  ██ (高精度)
      |  ██                                          ██
      |  ██                                          ██
  50% |  ██                                          ██
      |  ██             ░░ (低精度/失念)             ██
      |  ██             ░░                           ██
   0% +--------------------------------------------------->
       プロンプト先頭      プロンプト中央部          プロンプト末尾
       (Instructions)     (検索結果・長文ログ)       (最新の質問)

# 動作確認済みライブラリバージョン: tiktoken>=0.7.0, openai>=1.30.0
import os
import sys
from typing import List, Dict, Any, Optional
import tiktoken
from openai import OpenAI

class ContextManager:
    """Lost in the Middle 現象を回避し、トークン数を厳格に制御するコンテキストマネージャー"""
    
    def __init__(self, model_name: str = "gpt-4o", max_allowed_tokens: int = 2500):
        self.model_name = model_name
        self.max_allowed_tokens = max_allowed_tokens
        
        # モデルに対応する tiktoken エンコーダーの取得
        try:
            self.encoder = tiktoken.encoding_for_model(model_name)
        except KeyError:
            # 該当モデルが見つからない場合は cl100k_base をデフォルト使用
            self.encoder = tiktoken.get_encoding("cl100k_base")

    def count_tokens(self, text: str) -> int:
        """文字列の正確なトークン数を計算します"""
        return len(self.encoder.encode(text))

    def truncate_text_by_tokens(self, text: str, max_tokens: int) -> str:
        """指定したトークン数以内に文字列を正確に切り詰めます"""
        tokens = self.encoder.encode(text)
        if len(tokens) <= max_tokens:
            return text
        # トークン配列を切断してデコード復元
        truncated_tokens = tokens[:max_tokens]
        return self.encoder.decode(truncated_tokens) + "\n...[中略: トークン制限のため圧縮されました]..."

    def build_structured_context(
        self,
        system_instruction: str,
        user_profile: str,
        retrieved_documents: List[str],
        current_query: str
    ) -> List[Dict[str, str]]:
        """4層レイヤード構造（Prefix/Middle/Suffix）に基づいて最適なコンテキストメッセージ配列を生成します。"""
        
        # 1. 必須レイヤー (Layer 1, 2, 4) の固定トークン数を計算
        layer1_text = f"【システム絶対指示】\n{system_instruction.strip()}"
        layer2_text = f"【ユーザー定常記憶】\n{user_profile.strip()}"
        layer4_text = f"【最新の質問・実行指示】\n{current_query.strip()}\n\n回答は指示に従い論理的に述べてください。"

        token_l1 = self.count_tokens(layer1_text)
        token_l2 = self.count_tokens(layer2_text)
        token_l4 = self.count_tokens(layer4_text)

        reserved_tokens = token_l1 + token_l2 + token_l4
        
        print(f"📊 トークン計算: 固定レイヤー消費 = {reserved_tokens} トークン (上限 = {self.max_allowed_tokens})")

        # 2. Layer 3 (中央バッファ: RAGドキュメント) に割り当て可能な残容量を算出
        available_for_l3 = self.max_allowed_tokens - reserved_tokens

        if available_for_l3 <= 0:
            print("⚠️ 警告: 固定レイヤーのみで上限を超過しています。Layer 2をトリミングします。")
            layer2_text = self.truncate_text_by_tokens(layer2_text, max(50, self.max_allowed_tokens - token_l1 - token_l4))
            available_for_l3 = 0

        # 3. Layer 3 (Middle Buffer) の動的構築と圧縮
        l3_combined_text = ""
        if available_for_l3 > 0 and retrieved_documents:
            raw_l3_text = "【参照ナレッジドキュメント (RAG)】\n" + "\n---\n".join(retrieved_documents)
            l3_tokens = self.count_tokens(raw_l3_text)
            
            if l3_tokens > available_for_l3:
                print(f"✂️ Layer 3 が容量突破 ({l3_tokens} > {available_for_l3}) -> 動的トリミングを実行します。")
                l3_combined_text = self.truncate_text_by_tokens(raw_l3_text, available_for_l3)
            else:
                l3_combined_text = raw_l3_text
                print(f"✅ Layer 3 は容量内です ({l3_tokens} / {available_for_l3} トークン)")

        # 4. OpenAI / Anthropic 互換のメッセージ配列の組み立て
        #  - 先頭 (System): Layer 1 + Layer 2
        #  - 中央 (User): Layer 3 (ノード中央)
        #  - 末尾 (User): Layer 4 (思考直前の最高アテンション位置)
        
        messages = [
            {
                "role": "system",
                "content": f"{layer1_text}\n\n{layer2_text}"
            }
        ]

        if l3_combined_text:
            messages.append({
                "role": "user",
                "content": f"{l3_combined_text}"
            })

        messages.append({
            "role": "user",
            "content": f"{layer4_text}"
        })

        total_final_tokens = sum(self.count_tokens(m["content"]) for m in messages)
        print(f"🎉 最終構築プロンプト総トークン数: {total_final_tokens} トークン")

        return messages


# =====================================================================
# 実践デモ実行
# =====================================================================
def main():
    # OpenAI APIキーのセットアップ（環境変数がない場合はデモモード）
    api_key = os.environ.get("OPENAI_API_KEY", "dummy_key_for_testing")
    
    # マネージャーの初期化（上限をあえて小さめの 800 トークンに設定して挙動をテスト）
    manager = ContextManager(model_name="gpt-4o", max_allowed_tokens=800)

    # 1. 各レイヤーのダミーデータ作成
    sys_instruction = "あなたはエンタープライズソリューションアーキテクトです。質問に対して結論ファーストで回答してください。"
    user_profile = "ユーザー名: 山田太郎 | 役職: CTO | 関心技術: MCP, RAG, Python, Kubernetes"
    
    # あえて長大なダミーRAGドキュメントを作成（中間バッファ溢れの検証用）
    huge_documents = [
        f"ドキュメント #{i}: これはシステム仕様書の一部です。" + "詳細なパラメータ設定についての記述が続いています。" * 20
        for i in range(1, 10)
    ]
    
    query = "システムにMCPを導入した場合のトークンコスト削減効果とアーキテクチャ上のメリットを教えてください。"

    # 2. 層状コンテキストの自動組み立て
    print("==================================================")
    print("       Context Manager による構造化実行         ")
    print("==================================================")
    structured_messages = manager.build_structured_context(
        system_instruction=sys_instruction,
        user_profile=user_profile,
        retrieved_documents=huge_documents,
        current_query=query
    )

    print("\n--- 組み立てられたプロンプトのメッセージ構造 ---")
    for idx, msg in enumerate(structured_messages, 1):
        print(f"\n【メッセージ #{idx} - Role: {msg['role']}】")
        print(msg['content'][:300] + ("..." if len(msg['content']) > 300 else ""))

    # APIキーが設定されている場合は実際にLLMを呼び出し疎通
    if api_key != "dummy_key_for_testing":
        print("\n🤖 LLM (GPT-4o) への問い合わせを実行中...")
        client = OpenAI(api_key=api_key)
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=structured_messages,
            temperature=0.2
        )
        print("\n--- LLMからの返答 ---")
        print(response.choices[0].message.content)
    else:
        print("\n💡 NOTE: OPENAI_API_KEY が設定されていないため、LLM呼び出しはスキップされました。")

if __name__ == "__main__":
    main()

【Prompt Caching を最大化するレイヤー設計】
[Layer 1: System Instruction]   <-- 完全固定 (キャッシュ命中率 100%)
[Layer 2: User Profile/Memory]  <-- ユーザー単位固定 (キャッシュ命中)
----------------------------------- Cache Boundary (境界線) -----------------------------------
[Layer 3: Dynamic Documents]    <-- リクエストごとに変動
[Layer 4: Current Query]        <-- 毎回変動