OpenAI Responses API 400 invalid_output_text 해결 가이드

# (개념 예시) JSON schema로 출력 강제
resp = client.responses.create(
    model="gpt-4.1-mini",
    input="사용자 정보를 JSON으로 만들어줘",
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "user_profile",
            "schema": {
                "type": "object",
                "properties": {
                    "name": {"type": "string"},
                    "age": {"type": "integer"}
                },
                "required": ["name", "age"],
                "additionalProperties": False
            }
        }
    }
)

# 여기서 output_text만 기대하면 문제가 될 수 있음
print(resp.output_text)

from openai import OpenAI

client = OpenAI()

def extract_texts(response):
    texts = []
    for item in response.output:
        # SDK/버전에 따라 구조가 다를 수 있어 방어적으로 처리
        t = getattr(item, "type", None)
        if t in ("output_text", "text"):
            # 일부 SDK는 item.text, 일부는 item.content 등
            if hasattr(item, "text"):
                texts.append(item.text)
            elif hasattr(item, "content"):
                texts.append(item.content)
    return "\n".join(texts).strip()

resp = client.responses.create(
    model="gpt-4.1-mini",
    input="한 문장으로 요약해줘: HTTP 400은 클라이언트 요청 오류다."
)

print(extract_texts(resp))

from openai import OpenAI
import json

client = OpenAI()

# 예시 tool(실제로는 DB 조회/외부 API 호출 등)
def get_weather(city: str):
    return {"city": city, "temp_c": 3, "condition": "cloudy"}

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "도시의 현재 날씨를 가져옵니다",
            "parameters": {
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
                "additionalProperties": False,
            },
        },
    }
]

resp = client.responses.create(
    model="gpt-4.1-mini",
    input="서울 날씨 알려줘",
    tools=TOOLS,
)

# 1) tool 호출이 있는지 확인
for item in resp.output:
    if getattr(item, "type", None) in ("tool_call", "function_call"):
        fn = item.function.name
        args = json.loads(item.function.arguments)

        if fn == "get_weather":
            result = get_weather(**args)

            # 2) tool 결과를 다시 모델에 전달해 최종 텍스트 생성
            resp2 = client.responses.create(
                model="gpt-4.1-mini",
                input=[
                    {"role": "user", "content": "서울 날씨 알려줘"},
                    {"role": "tool", "name": fn, "content": json.dumps(result)},
                ],
            )
            # 최종 텍스트 추출
            print(getattr(resp2, "output_text", ""))

def safe_output_text(resp) -> str:
    # SDK에서 제공하는 output_text가 있으면 우선 사용
    text = getattr(resp, "output_text", None)
    if isinstance(text, str) and text.strip():
        return text

    # 없으면 output 배열에서 최대한 찾아본다
    if hasattr(resp, "output"):
        chunks = []
        for item in resp.output:
            if getattr(item, "type", None) in ("output_text", "text"):
                if hasattr(item, "text"):
                    chunks.append(item.text)
        return "\n".join(chunks).strip()

    return ""  # text가 없는 응답도 정상 케이스로 취급

import json

def dump_response(resp):
    # SDK 객체를 dict로 바꾸는 방법은 버전에 따라 다릅니다.
    # 가능한 경우 model_dump_json / to_dict 등을 사용하세요.
    if hasattr(resp, "model_dump"):
        print(json.dumps(resp.model_dump(), ensure_ascii=False, indent=2))
    elif hasattr(resp, "to_dict"):
        print(json.dumps(resp.to_dict(), ensure_ascii=False, indent=2))
    else:
        # 최후의 수단
        print(resp)

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from openai import OpenAI

app = FastAPI()
client = OpenAI()

class Req(BaseModel):
    prompt: str

def extract_any_text(resp) -> str:
    text = getattr(resp, "output_text", None)
    if isinstance(text, str) and text.strip():
        return text

    if hasattr(resp, "output"):
        parts = []
        for item in resp.output:
            t = getattr(item, "type", None)
            if t in ("output_text", "text"):
                if hasattr(item, "text") and item.text:
                    parts.append(item.text)
        return "\n".join(parts).strip()

    return ""

@app.post("/ask")
def ask(req: Req):
    try:
        resp = client.responses.create(
            model="gpt-4.1-mini",
            input=req.prompt,
            # tools/response_format 등을 켰다면 여기서부터 text-only 가정이 깨질 수 있음
        )
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"upstream error: {e}")

    text = extract_any_text(resp)
    if not text:
        # 여기서 바로 500으로 터뜨리기보다,
        # output 전체를 로깅하고 422/204 등 정책적으로 처리하는 게 운영에 유리
        raise HTTPException(status_code=422, detail="no text output (maybe tool/json output)")

    return {"answer": text}