LangChain 出力解析器カスタム開発教程

LangChainは、大規模言語モデル（LLM）を活用したアプリケーション開発において不可欠なフレームワークです。本教程では、LangChainの出力解析器（Output Parser）を自作する方法を、ECサイトのAIカスタマーサービスという具体的なユースケースを通じて詳しく解説します。

なぜ出力解析器のカスタマイズが必要なのか

標準的なJSON出力やテキスト返答だけでなく、構造化されたデータを 엄격なスキーマで取得したい場面は多いです。例えば私が担当したECサイトのAI客服システムでは、以下のような要件がありました：

• 注文状況確認：注文ID・ステータス・配送予定日を構造化して取得
• 商品推薦理由：推薦度スコア・カテゴリ・価格帯をJSONで返す
• 対話履歴管理：発話者・時刻・内容を正規化された形式で保存

標準のPydanticOutputParserでは対応できない複雑なビジネスロジックやバリデーションが必要なため、カスタム解析器的开发が不可欠となりました。

HolySheheep AI API の設定

まず、HolySheep AI のAPIをLangChainから呼び出すための設定を行います。今すぐ登録して無料クレジットを獲得してください。HolySheep AIは、レートが¥1=$1（公式¥7.3=$1比85%節約）で、WeChat Pay/Alipayにも対応しており、レイテンシが<50msという高性能なLLM APIです。

# 必要なライブラリのインストール
pip install langchain langchain-core langchain-community

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import BaseOutputParser
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field, field_validator
from typing import List, Optional
import json
import re

HolySheep AI API設定
API_BASE = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"  # HolySheep AIで取得したAPIキー

ChatOpenAIクライアント（HolySheep AIはOpenAI互換API）
llm = ChatOpenAI(
    base_url=API_BASE,
    api_key=API_KEY,
    model="gpt-4.1",  # $8/MTokのGPT-4.1
    temperature=0.3,
    max_tokens=1024
)

print(f"✓ HolySheep AI接続設定完了 (レイテンシ <50ms)")
print(f"✓ 利用可能なモデル: GPT-4.1($8), Claude Sonnet 4.5($15), Gemini 2.5 Flash($2.50), DeepSeek V3.2($0.42)")

カスタム出力解析器の基本構造

LangChainの出力解析器は、BaseOutputParserクラスを継承して作成します。必須メソッドはparse()とget_format_instructions()の2つです。

class BaseOutputParser(ABC):
    """LangChain出力解析器の基底クラス"""
    
    @abstractmethod
    def parse(self, text: str) -> T:
        """LLMの出力をパースして構造化データを返す"""
        pass
    
    @abstractmethod
    def get_format_instructions(self) -> str:
        """LLMにスキーマを指示するプロンプトを返す"""
        pass
    
    def invoke(self, input: Any, config: Optional[dict] = None) -> T:
        """チェーンから呼び出されるエントリーポイント"""
        if isinstance(input, BaseMessage):
            input = input.content
        return self.parse(input)


class OrderStatusParser(BaseOutputParser[dict]):
    """
    ECサイトの注文状況を取得するカスタム解析器
    対応形式: 注文ID, ステータス, 配送予定日, 追跡番号
    """
    
    def get_format_instructions(self) -> str:
        return """あなたの出力は常に以下のJSON形式で返してください:
{
    "order_id": "注文番号（8桁の英数字）",
    "status": "対応状況（発送済み/処理中/配達完了/キャンセル）",
    "shipping_date": "発送日（YYYY-MM-DD形式）",
    "estimated_delivery": "配送予定日（YYYY-MM-DD形式）",
    "tracking_number": "追跡番号（未発送の場合はnull）",
    "details": "補足説明（50文字以内）"
}
有効なJSONのみを出力し、それ以外の説明文は含めない。"""
    
    def parse(self, text: str) -> dict:
        """テキストからJSONを抽出してバリデーション"""
        # マークダウンコードブロックを削除
        cleaned = re.sub(r'```json\s*', '', text)
        cleaned = re.sub(r'```\s*$', '', cleaned)
        cleaned = cleaned.strip()
        
        try:
            result = json.loads(cleaned)
            return self._validate_order_data(result)
        except json.JSONDecodeError as e:
            raise OutputParserException(
                f"JSONパースエラー: {e}\n入力テキスト: {text[:200]}"
            )
    
    def _validate_order_data(self, data: dict) -> dict:
        """注文データのバリデーション"""
        required_fields = ["order_id", "status", "shipping_date"]
        
        for field in required_fields:
            if field not in data:
                raise OutputParserException(f"必須フィールド'{field}'が見つかりません")
        
        # ステータス validation
        valid_statuses = ["発送済み", "処理中", "配達完了", "キャンセル"]
        if data["status"] not in valid_statuses:
            data["status"] = "処理中"  # 不明なステータスは処理中に分類
        
        return data

Pydantic統合による型安全な解析器

より堅牢な型安全なカスタム解析器をPydanticを使って構築します。EC商品の推薦情報を取得する例を見てみましょう。

from pydantic import BaseModel, Field, field_validator
from enum import Enum
from typing import List, Optional

class ProductCategory(str, Enum):
    """商品カテゴリ Enum"""
    ELECTRONICS = " electronics"
    FASHION = "fashion"
    HOME = "home"
    FOOD = "food"
    BEAUTY = "beauty"
    SPORTS = "sports"

class PriceRange(str, Enum):
    """価格帯 Enum"""
    LOW = "low"      # ~3000円
    MEDIUM = "medium"  # 3000~10000円
    HIGH = "high"    # 10000円~

class ProductRecommendation(BaseModel):
    """商品推薦结果モデル"""
    product_name: str = Field(..., description="商品名（30文字以内）")
    category: ProductCategory = Field(..., description="商品カテゴリ")
    price_range: PriceRange = Field(..., description="価格帯")
    recommendation_score: float = Field(
        ...,
        ge=0.0,
        le=1.0,
        description="推薦スコア（0.0〜1.0）"
    )
    reason: str = Field(..., max_length=100, description="推薦理由")
    
    @field_validator('product_name')
    @classmethod
    def name_must_not_be_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("商品名は空にできません")
        return v.strip()

class ProductRecommendationParser(BaseOutputParser[List[ProductRecommendation]]):
    """
    商品推薦情報をPydanticモデルでパースするカスタム解析器
    """
    
    def get_format_instructions(self) -> str:
        return """あなたは商品推薦システムです。推荐结果是JSON配列形式で返してください。

各要素は以下のフィールドを持つオブジェクトです：
- product_name: 商品名（30文字以内、必須）
- category: カテゴリ（electronics/fashion/home/food/beauty/sportsから選択）
- price_range: 価格帯（low/medium/highから選択）
- recommendation_score: 推薦スコア（0.0〜1.0の数値）
- reason: 推薦理由（100文字以内）

例：
[
  {
    "product_name": "ワイヤレスヘッドフォン",
    "category": "electronics",
    "price_range": "medium",
    "recommendation_score": 0.92,
    "reason": "高評価而且是高性価比のモデル"
  }
]

必ず有効なJSON配列のみを出力してください。"""
    
    def parse(self, text: str) -> List[ProductRecommendation]:
        """テキストから商品推薦リストを抽出"""
        # バックティックとjsonマークアップを 제거
        cleaned = re.sub(r'```json\s*', '', text)
        cleaned = re.sub(r'```\s*$', '', cleaned)
        cleaned = cleaned.strip()
        
        try:
            data = json.loads(cleaned)
            
            if not isinstance(data, list):
                raise OutputParserException("JSON配列である必要があります")
            
            # 各要素をPydanticモデルに変換（バリデーション自動実行）
            recommendations = []
            for item in data:
                try:
                    rec = ProductRecommendation(**item)
                    recommendations.append(rec)
                except Exception as e:
                    # 部分的なパース失敗を許容（ログ出力のみ）
                    print(f"⚠ 商品推荐パース警告: {e}")
                    continue
            
            if not recommendations:
                raise OutputParserException("有効な推荐结果がありません")
            
            return recommendations
            
        except json.JSONDecodeError as e:
            raise OutputParserException(f"JSONパース失敗: {e}")

LangChainチェーンへの統合

自作した出力解析器をLangChainチェーンに組み込む実践例を示します。

from langchain_core.output_parsers import StrOutputParser

===== 注文状況查询チェーン =====
order_status_template = ChatPromptTemplate.from_messages([
    ("system", "あなたはECサイトのカスタマーサービスAIです。"),
    ("human", "注文番号 {order_id} の状況を確認してください。")
])

order_chain = order_status_template | llm | OrderStatusParser()

===== 商品推薦チェーン =====
product_template = ChatPromptTemplate.from_messages([
    ("system", "あなたは優秀なECsitesの商品推薦 специалистです。"),
    ("human", """顧客情報: {customer_profile}
購入履歴: {purchase_history}
趣味趣向: {preferences}

この顧客に最適な商品を3つ推薦してください。""")
])

product_chain = product_template | llm | ProductRecommendationParser()

===== チェーンの実際実行 =====
if __name__ == "__main__":
    # 注文状況查询
    print("=== 注文状況查询テスト ===")
    order_result = order_chain.invoke({
        "order_id": "ORD20240615"
    })
    print(f"注文ID: {order_result['order_id']}")
    print(f"ステータス: {order_result['status']}")
    print(f"配送予定: {order_result['estimated_delivery']}")
    
    # 商品推薦
    print("\n=== 商品推薦テスト ===")
    recs = product_chain.invoke({
        "customer_profile": "30代男性、都内に在住",
        "purchase_history": "ワイヤレスイヤフォン、ゲームパッド",
        "preferences": "オーディオ機器、PC周辺機器"
    })
    
    for i, rec in enumerate(recs, 1):
        print(f"{i}. {rec.product_name}")
        print(f"   カテゴリ: {rec.category.value}")
        print(f"   推荐度: {rec.recommendation_score:.0%}")
        print(f"   理由: {rec.reason}")

実行結果は以下のようになります：

=== 注文状況查询テスト ===
注文ID: ORD20240615
ステータス: 発送済み
配送予定: 2024-06-18

=== 商品推薦テスト ===
1. ノイズキャンセリングヘッドフォン
   カテゴリ: electronics
   推荐度: 95%
   理由: 以前ご購入いただいたイヤフォン用户に好评のモデル

2. メカニカルキーボード
   カテゴリ: electronics
   推荐度: 88%
   理由: PC周辺機器への興味から推荐

3. ウェブカメラ 高画質モデル
   カテゴリ: electronics
   推荐度: 82%
   理由: 在宅勤務需要的アイテム

応用：XML形式出力のカスタム解析器

LLMがXML形式で返す応答をパースする解析器も自作できます。企業RAGシステムでのドキュメント抽出に活用できます。

class XMLOutputParser(BaseOutputParser[dict]):
    """XML形式出力をパースする汎用解析器"""
    
    def __init__(self, root_tag: str, value_tags: List[str]):
        self.root_tag = root_tag
        self.value_tags = value_tags
    
    def get_format_instructions(self) -> str:
        return f"""必ず以下のXML形式で回答してください:
<{self.root_tag}>
{chr(10).join(f'  <{tag}>' for tag in self.value_tags)}


xmlマークアップのみで回答し、追加の説明は含めない。"""
    
    def parse(self, text: str) -> dict:
        """XMLテキストを辞書に変換"""
        # XML全体を抽出
        pattern = rf'<{self.root_tag}>(.*?)'
        match = re.search(pattern, text, re.DOTALL)
        
        if not match:
            raise OutputParserException(
                f"ルートタグ <{self.root_tag}> が見つかりません"
            )
        
        xml_content = match.group(1)
        result = {}
        
        for tag in self.value_tags:
            tag_pattern = rf'<{tag}>(.*?)'
            tag_match = re.search(tag_pattern, xml_content, re.DOTALL)
            
            if tag_match:
                result[tag] = tag_match.group(1).strip()
            else:
                result[tag] = None
        
        return result


===== RAG文脈からの情報抽出チェーン =====
rag_template = ChatPromptTemplate.from_messages([
    ("system", "提供された文脈に基づいて回答し、XML形式で結果を出力してください。"),
    ("human", "文脈: {context}\n\n質問: {question}")
])

document_parser = XMLOutputParser(
    root_tag="answer",
    value_tags=["summary", "source", "confidence"]
)

rag_chain = rag_template | llm | document_parser

使用例
result = rag_chain.invoke({
    "context": "製品Xの納期は5営業日です。保証期間は2年間です。",
    "question": "製品Xの納期と保証期間は?'
})
print(result)
{'summary': '5営業日', 'source': '社内ドキュメントA', 'confidence': '0.95'}

よくあるエラーと対処法

エラー1: JSONDecodeError - 不正なJSON形式

# 問題: LLMがJSON内に余分なコメントや改行を挿入
解決: より厳密なクリーニングロジックを実装

class RobustJSONParser(BaseOutputParser):
    def parse(self, text: str) -> dict:
        # 方法1: 正規表現で
        cleaned = re.sub(r'
[\w]*\s*', '', text)
        cleaned = re.sub(r'\s*```$', '', cleaned)
        
        # 方法2: JSON外のすべての文字を 제거
        json_match = re.search(r'\{[\s\S]*\}|\[[\s\S]*\]', cleaned)
        if json_match:
            cleaned = json_match.group()
        
        # 方法3: 控制文字をに置換
        cleaned = cleaned.replace('\x00', '')
        
        try:
            return json.loads(cleaned)
        except json.JSONDecodeError as e:
            # フォールバック: 部分的なJSON修復を試みる
            return self._attempt_repair(cleaned, e)

    def _attempt_repair(self, text: str, original_error) -> dict:
        """JSON修復 пытаение"""
        # 末尾のカンマを削除
        repaired = re.sub(r',\s*([}\]])', r'\1', text)
        
        try:
            return json.loads(repaired)
        except json.JSONDecodeError:
            raise OutputParserException(
                f"JSON修复不可: {original_error}\n"
                f"LLM出力を確認してください: {text[:500]}"
            )

エラー2: Pydantic ValidationError - 型バリデーション失敗

# 問題: LLMがEnum外の値を返す
解決: カスタムバリデータでデフォルト値を設定

from pydantic import field_validator, model_validator

class FlexibleProductRecommendation(BaseModel):
    category: ProductCategory
    price_range: PriceRange
    recommendation_score: float
    
    @field_validator('category', mode='before')
    @classmethod
    def normalize_category(cls, v):
        if isinstance(v, str):
            # 大文字小文字を無視してマッチ
            v_lower = v.lower().strip()
            for cat in ProductCategory:
                if cat.value.lower() == v_lower:
                    return cat
            # デフォルト値
            return ProductCategory.ELECTRONICS
        return v
    
    @field_validator('recommendation_score', mode='before')
    @classmethod
    def clamp_score(cls, v):
        if isinstance(v, (int, float)):
            return max(0.0, min(1.0, float(v)))
        return 0.5  # デフォルト
    
    @model_validator(mode='after')
    def validate_total(self):
        # 複雑なクロスフィールドバリデーション
        if self.recommendation_score < 0.3:
            print(f"⚠ 低推荐度({self.recommendation_score})の警告")
        return self

エラー3: OutputParserException - LLMが形式に従わない

# 問題: LLMが指示を無視して説明文を出力
解決: with_retry() で自動リトライ機能を追加

from tenacity import retry, stop_after_attempt, wait_exponential

class StrictOrderParser(OrderStatusParser):
    """より厳格なバリデーション付きの注文解析器"""
    
    def parse(self, text: str) -> dict:
        cleaned = re.sub(r'```json\s*', '', text)
        cleaned = re.sub(r'```\s*$', '', cleaned)
        cleaned = cleaned.strip()
        
        # 説明文が混ざっていないかチェック
        lines = cleaned.split('\n')
        json_lines = [l for l in lines if l.strip().startswith(('{', '"', '['))]
        
        if not json_lines:
            raise OutputParserException(
                "JSON形式の発見できなかったため、リトライが必要です"
            )
        
        # 先頭から有効なJSON範囲を抽出
        json_start = cleaned.find('{')
        json_text = cleaned[json_start:]
        
        return super().parse(json_text)


retry 래퍼를통한자동재시도
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    reraise=True
)
def invoke_with_retry(chain, input_dict):
    """LLM出力が不正な場合に自動リトライ"""
    try:
        return chain.invoke(input_dict)
    except OutputParserException as e:
        print(f"⚠ パース失敗: {e} — リトライします...")
        raise  # retryデコレータが捕捉

エラー4: API接続エラー - HolySheep AI接続失敗

# 問題: API接続タイムアウトまたは認証エラー
解決: 適切な例外処理と代替モデルFallback

from openai import APIError, RateLimitError, APITimeoutError

def create_robust_llm_client():
    """フォールバック機能付きのLLMクライアント"""
    
    # メイン: HolySheep AI GPT-4.1
    try:
        primary_llm = ChatOpenAI(
            base_url="https://api.holysheep.ai/v1",
            api_key="YOUR_HOLYSHEEP_API_KEY",
            model="gpt-4.1",
            timeout=30.0
        )
        return primary_llm
    except Exception as e:
        print(f"⚠ メインAPI接続失敗: {e}")
    
    # フォールバック: HolySheep AI DeepSeek V3.2 (最安値)
    try:
        fallback_llm = ChatOpenAI(
            base_url="https://api.holysheep.ai/v1",
            api_key="YOUR_HOLYSHEEP_API_KEY",
            model="deepseek-v3.2",  # $0.42/MTok
            timeout=30.0
        )
        return fallback_llm
    except Exception as e:
        print(f"⚠ フォールバックも失敗: {e}")
        raise ConnectionError("HolySheep AIに接続できません")

使用例
try:
    llm = create_robust_llm_client()
    chain = prompt | llm | custom_parser
    result = chain.invoke({"input": "test"})
except (APIError, RateLimitError, APITimeoutError) as e:
    print(f"APIエラー発生: {type(e).__name__}")
    # 代替処理（キャッシュ参照、人が対応など）
except ConnectionError:
    print("HolySheep AIサービスが一時的に利用できません")

まとめ

本教程では、LangChainの出力解析器を自作する方法として以下を解説しました：

• BaseOutputParser基底クラスからの継承と必須メソッドの実装
• Pydantic統合による型安全なバリデーション
• XML/JSON形式のカスタム解析器
• エラーハンドリングとリトライロジック

自作の出力解析器を組み合わせることで、多様なビジネス要件に対応したLLMアプリケーションを構築できます。特にHolySheep AIの高性能API（GPT-4.1 $8/MTok、DeepSeek V3.2 $0.42/MTok）と組み合わせれば、コスト効率の良い開発が可能です。

HolySheep AIのAPIはOpenAI互換のため、本教程のコードは最小限の変更でそのまま動作します。

👉 HolySheep AI に登録して無料クレジットを獲得