Gemma 4構造化出力：毎回信頼できるJSONを取得する方法

# これが欲しい：
{"sentiment": "positive", "confidence": 0.92, "topics": ["pricing", "support"]}

# これは欲しくない：
"The sentiment of this text is positive, with a confidence of about 92%..."

import requests
import json

response = requests.post("http://localhost:11434/api/chat", json={
    "model": "gemma4:26b",
    "messages": [
        {
            "role": "system",
            "content": """You are a JSON-only response API. 
You MUST respond with valid JSON and nothing else.
No markdown, no explanation, no code blocks — just raw JSON.

Schema:
{
  "sentiment": "positive" | "negative" | "neutral",
  "confidence": number between 0 and 1,
  "topics": string[],
  "summary": string (one sentence)
}"""
        },
        {
            "role": "user",
            "content": "Analyze: 'The new update is amazing! The UI is so much cleaner and everything loads faster. Only complaint is the price went up.'"
        }
    ],
    "stream": False,
})

result = json.loads(response.json()["message"]["content"])
print(result)

response = requests.post("http://localhost:11434/api/chat", json={
    "model": "gemma4:26b",
    "messages": [
        {
            "role": "system",
            "content": "Analyze the sentiment of the given text. Return: sentiment (positive/negative/neutral), confidence (0-1), topics (list), summary (one sentence)."
        },
        {
            "role": "user",
            "content": "The customer service was terrible but the product itself is excellent."
        }
    ],
    "format": "json",
    "stream": False,
})

# これは有効なJSONが保証されている
result = response.json()["message"]["content"]
parsed = json.loads(result)

from pydantic import BaseModel, Field
from typing import Literal
import json
import requests

class SentimentResult(BaseModel):
    sentiment: Literal["positive", "negative", "neutral"]
    confidence: float = Field(ge=0, le=1)
    topics: list[str]
    summary: str

def analyze_sentiment(text: str) -> SentimentResult:
    schema_str = json.dumps(SentimentResult.model_json_schema(), indent=2)
    
    response = requests.post("http://localhost:11434/api/chat", json={
        "model": "gemma4:26b",
        "messages": [
            {
                "role": "system",
                "content": f"""Respond with JSON matching this exact schema:
{schema_str}

No other text. Just valid JSON."""
            },
            {
                "role": "user",
                "content": f"Analyze this text: {text}"
            }
        ],
        "format": "json",
        "stream": False,
    })
    
    raw = json.loads(response.json()["message"]["content"])
    return SentimentResult.model_validate(raw)

# 使用
result = analyze_sentiment("Great product, terrible shipping time.")
print(f"Sentiment: {result.sentiment} ({result.confidence:.0%})")
print(f"Topics: {', '.join(result.topics)}")

from pydantic import BaseModel, ValidationError
import json
import requests
import time

def get_structured_output(
    prompt: str,
    schema_class: type[BaseModel],
    model: str = "gemma4:26b",
    max_retries: int = 3,
) -> BaseModel:
    schema_str = json.dumps(schema_class.model_json_schema(), indent=2)
    
    for attempt in range(max_retries):
        try:
            response = requests.post("http://localhost:11434/api/chat", json={
                "model": model,
                "messages": [
                    {
                        "role": "system",
                        "content": f"Respond ONLY with JSON matching this schema:\n{schema_str}"
                    },
                    {"role": "user", "content": prompt}
                ],
                "format": "json",
                "stream": False,
                "options": {
                    "temperature": 0.1 if attempt == 0 else 0.3,
                },
            })
            
            raw = json.loads(response.json()["message"]["content"])
            return schema_class.model_validate(raw)
            
        except (json.JSONDecodeError, ValidationError) as e:
            if attempt == max_retries - 1:
                raise ValueError(
                    f"Failed to get valid output after {max_retries} attempts: {e}"
                )
            time.sleep(0.5)
    
    raise ValueError("Unreachable")

# 使用
class ProductReview(BaseModel):
    rating: int = Field(ge=1, le=5)
    pros: list[str]
    cons: list[str]
    recommendation: bool

review = get_structured_output(
    "Review: 'Solid laptop, great keyboard, battery could be better. 4/5 would buy again.'",
    ProductReview,
)

```json
{"key": "value"}
```​

def clean_json(text: str) -> str:
    text = text.strip()
    if text.startswith("```"):
        text = text.split("\n", 1)[1]  # 最初の行を削除
        text = text.rsplit("```", 1)[0]  # 最後のバッククォートを削除
    return text.strip()

class Result(BaseModel):
    name: str
    email: str | None = None  # モデルがメールを見つけられない可能性
    topics: list[str] = []    # デフォルトで空リスト

class Address(BaseModel):
    city: str
    country: str

class Person(BaseModel):
    name: str
    age: int
    address: Address  # 1レベルのネスト — 問題なし