🎯
Instructor: LLMからタイプセーフな構造化データを取得する仕組みとLiteLLMでの導入方法
はじめに
LLM（Large Language Model）から構造化されたデータを取得する際、JSONフォーマットでの出力を期待しても、実際には不完全な形式や型の不一致が発生することがあります。
Instructorは、この問題を解決するPythonライブラリで、Pydanticの型検証と自動リトライメカニズムを組み合わせることで、LLMからタイプセーフな構造化データを高い信頼性で取得できます。
本記事では、Instructorがどのようにしてタイプセーフを実現しているのか、その内部メカニズムをコードベースで解説し、LiteLLMを使った実装方法を紹介します。
Instructorの内部メカニズム
1. 基本的なアーキテクチャ
Instructorの中核は、以下の3つのコンポーネントで構成されています：
- Pydanticによる型定義と検証
- 自動リトライメカニズム
- エラーフィードバックによる修正
フロー図
2. リトライメカニズムの詳細実装
instructor/core/retry.pyのコードを見ると、リトライ処理の実装が確認できます：
def retry_sync(
func: Callable[T_ParamSpec, T_Retval],
response_model: type[T_Model] | None,
args: Any,
kwargs: Any,
max_retries: int | Retrying = 1,
...
) -> T_Model | None:
"""同期的なリトライ処理"""
# Tenacityライブラリを使用したリトライ設定
max_retries = initialize_retrying(max_retries, is_async=False)
try:
for attempt in max_retries:
with attempt:
try:
# LLMへのリクエスト実行
response = func(*args, **kwargs)
# レスポンスの処理と検証
return process_response(
response=response,
response_model=response_model,
validation_context=context,
strict=strict,
mode=mode,
)
except (ValidationError, JSONDecodeError) as e:
# 検証エラーの場合、エラー情報を次回リクエストに含める
kwargs = handle_reask_kwargs(
kwargs=kwargs,
mode=mode,
response=response,
exception=e,
)
raise e
3. エラーフィードバックの仕組み
handle_reask_kwargs関数は、検証エラーをLLMにフィードバックする処理を担当します。プロバイダーごとに異なる戦略を持っています：
def handle_reask_kwargs(
kwargs: dict[str, Any],
mode: Mode,
response: Any,
exception: Exception,
) -> dict[str, Any]:
"""検証エラーを次のリクエストに含める"""
# プロバイダーごとのリトライハンドラー
REASK_HANDLERS = {
Mode.TOOLS: reask_tools,
Mode.JSON: reask_md_json,
Mode.ANTHROPIC_TOOLS: reask_anthropic_tools,
Mode.MISTRAL_TOOLS: reask_mistral_tools,
# ... 他のプロバイダー
}
return REASK_HANDLERS[mode](kwargs, response, exception)
OpenAIの場合の実装例
instructor/providers/openai/utils.py：
def reask_tools(
kwargs: dict[str, Any],
response: Any,
exception: Exception,
):
"""OpenAIのツールモードでのリトライ処理"""
# エラーメッセージを含むツールレスポンスを作成
reask_msgs = [dump_message(response.choices[0].message)]
for tool_call in response.choices[0].message.tool_calls:
reask_msgs.append({
"role": "tool",
"tool_call_id": tool_call.id,
"name": tool_call.function.name,
"content": (
f"Validation Error found:\n{exception}\n"
"Recall the function correctly, fix the errors"
),
})
# メッセージ履歴にエラー情報を追加
kwargs["messages"].extend(reask_msgs)
return kwargs
この実装により、LLMは前回の失敗を理解し、次回はより正確な構造化データを生成できます。
4. Pydanticによる型検証
from_responseメソッドで、各プロバイダーのレスポンスをPydanticモデルに変換します：
@classmethod
def from_response(
cls,
completion: ChatCompletion,
validation_context: Optional[dict[str, Any]] = None,
strict: Optional[bool] = None,
mode: Mode = Mode.TOOLS,
) -> BaseModel:
"""レスポンスからPydanticモデルを生成"""
# プロバイダーごとのパース処理
if mode == Mode.ANTHROPIC_TOOLS:
return cls.parse_anthropic_tools(completion, validation_context, strict)
if mode in {Mode.TOOLS, Mode.MISTRAL_TOOLS, Mode.CEREBRAS_TOOLS}:
return cls.parse_tools(completion, validation_context, strict)
if mode in {Mode.JSON, Mode.JSON_SCHEMA, Mode.MD_JSON}:
return cls.parse_json(completion, validation_context, strict)
# ... 他のモード
LiteLLMでの導入方法
1. インストール
pip install "instructor[litellm]"
2. 基本的な使い方
シンプルな例
from litellm import completion
import instructor
from pydantic import BaseModel
# Instructorクライアントを作成
client = instructor.from_litellm(completion)
# 構造化データのスキーマを定義
class User(BaseModel):
name: str
age: int
email: str
# LLMから構造化データを取得
user = client.chat.completions.create(
model="gpt-3.5-turbo", # またはLiteLLM対応の任意のモデル
messages=[
{
"role": "user",
"content": "Extract: John Doe is 30 years old, email: john@example.com"
}
],
response_model=User,
max_retries=3, # 自動リトライ回数
)
print(user)
# User(name='John Doe', age=30, email='john@example.com')
非同期版
from litellm import acompletion
import instructor
import asyncio
from pydantic import BaseModel
# 非同期クライアントを作成
client = instructor.from_litellm(acompletion)
class Product(BaseModel):
name: str
price: float
in_stock: bool
async def extract_product():
product = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[
{
"role": "user",
"content": "MacBook Pro M3 costs $1999 and is currently available"
}
],
response_model=Product,
)
return product
# 実行
product = asyncio.run(extract_product())
print(product)
# Product(name='MacBook Pro M3', price=1999.0, in_stock=True)
3. LiteLLMの統合メカニズム
instructor/core/client.pyのfrom_litellm関数を見ると、統合の仕組みがわかります：
def from_litellm(
completion: Callable[..., Any] | Callable[..., Awaitable[Any]],
mode: instructor.Mode = instructor.Mode.TOOLS,
**kwargs: Any,
) -> Instructor | AsyncInstructor:
"""LiteLLMのcompletion関数をラップ"""
# 非同期関数かどうかを判定
is_async = inspect.iscoroutinefunction(completion)
if not is_async:
# 同期版のInstructorクライアントを返す
return Instructor(
client=None,
create=instructor.patch(create=completion, mode=mode),
mode=mode,
**kwargs,
)
else:
# 非同期版のAsyncInstructorクライアントを返す
return AsyncInstructor(
client=None,
create=instructor.patch(create=completion, mode=mode),
mode=mode,
**kwargs,
)
この実装により、LiteLLMのcompletion関数がInstructorの機能（型検証、リトライ、エラーフィードバック）でラップされます。
制約・前提条件
パフォーマンスとコスト
- リトライによる遅延: max_retries=3の場合、最大で元の3倍の時間とコストが発生
- トークン使用量: エラー履歴がメッセージに蓄積されるため、リトライ回数に応じて使用量が増加
- レート制限: プロバイダーごとのAPI制限に注意（特にリトライ時）
セキュリティ考慮事項
import os
from litellm import completion
# 環境変数でAPIキー管理
os.environ["OPENAI_API_KEY"] = "your_api_key_here"
os.environ["ANTHROPIC_API_KEY"] = "your_api_key_here"
# データプライバシー: 機密情報の扱いに注意
class SafeUserInfo(BaseModel):
# 機密情報は含めない設計
display_name: str = Field(description="Public display name only")
age_range: str = Field(description="Age range like '20-30', not exact age")
プロダクション環境での推奨設定
# 本番環境向けの保守的な設定
client = instructor.from_litellm(
completion,
mode=instructor.Mode.TOOLS # 安定性重視
)
result = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=messages,
response_model=YourModel,
max_retries=2, # コスト制御
timeout=15, # 応答時間制限
temperature=0.1 # 安定性重視
)
まとめ
Instructorは以下の仕組みでタイプセーフな構造化データの取得を実現しています：
- Pydanticによる厳密な型定義と検証
- 検証エラー時の自動リトライ（ただし、コスト・時間の増加に注意）
- エラー内容のLLMへのフィードバック
- プロバイダーごとに最適化されたエラーハンドリング
LiteLLMとの組み合わせにより、複数のLLMプロバイダーを統一的に扱いながら、高い信頼性で構造化データを取得できます。
ただし、100%の成功保証はない点に留意し、適切なエラーハンドリング、コスト管理、セキュリティ対策を実装することが重要です。
Discussion