OpenAI 함수 호출 테스트
openai package를 사용하는 테스트 코드 작성할거야. 목표: 2가지 모델에 대해 2가지 api type(completions, responses)에 대해 function calling 테스트를 하고 싶어. 총 4개 조합에 대해 설정할 수 있게 코드 작성 하자. 툴은 간단히 특정 폴더의 파일 조회, 특정 폴더의 파일 읽기 두 가지 툴을 구현할거야 2개 모델에 대한 엔드포인트, api 키, 모델명을 코드 내에서 설정할 수 있도록 할 것.
좋아! 아래는 “두 모델 × 두 API 타입(completions, responses) × 함수 호출(function calling)” 조합을 한 파일에서 모두 테스트할 수 있는 파이썬 스크립트야.
요구사항 정리대로
-
두 가지 파일 관련 툴(폴더 내 파일 목록 조회, 파일 읽기)
-
2개 모델에 대한 엔드포인트·API 키·모델명을 코드 내에서 설정 가능
-
4가지 조합을 원하는 대로 실행 가능
을 모두 담았어. 최신 OpenAI Python SDK의 사용법과 Responses API의 툴 호출 흐름은 공식 문서의 구조를 따랐어. 참고: function calling과 tools 정의 방식은 공식 가이드와 동일하고, Chat Completions API에서는 tool_calls를 message에, Responses API에서는 output 항목과 tool_calls를 통해 주고받는 형태야. 관련 문서: Function Calling 가이드와 Responses API 레퍼런스. OpenAI 플랫폼+1
스크립트는 다음처럼 동작해
-
MODELS 섹션에서 model A/B의 endpoint, api_key, model 이름을 채워둔다(예시 포함).
-
TOOL 정의는 공통(JSON Schema).
-
run_with_chat_completions(…), run_with_responses(…) 두 함수가 각각 함수 호출 루프를 수행.
-
main에서 네 가지 조합을 순차 실행하거나, 원하는 조합만 선택 실행하도록 CLI 옵션 제공.
코드 복사해서 test_openai_tools.py로 저장한 뒤, samples 폴더에 테스트용 파일을 몇 개 넣고 돌려봐.
# test_openai_tools.py
# Python 3.10+
# pip install --upgrade openai
import os
import json
import argparse
from dataclasses import dataclass
from typing import Dict, Any, List, Optional
from openai import OpenAI
# =========================
# 1) 구성(모델/엔드포인트/키)
# =========================
@dataclass
class ModelConfig:
name: str
base_url: str # ex) "https://api.openai.com/v1"
api_key: str # ex) os.getenv("OPENAI_API_KEY")
# 여기서 두 모델에 대한 엔드포인트/키/모델명을 설정하세요.
# 실제 키를 코드에 하드코딩하지 말고, 환경변수를 권장합니다.
MODELS = {
"A": ModelConfig(
name="gpt-4.1-mini", # 예시 (원하는 최신 모델명으로 변경)
base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
api_key=os.getenv("OPENAI_API_KEY", "YOUR_KEY_FOR_MODEL_A")
),
"B": ModelConfig(
name="gpt-4o-mini", # 예시 (원하는 최신 모델명으로 변경)
base_url=os.getenv("OPENAI_BASE_URL_B", "https://api.openai.com/v1"),
api_key=os.getenv("OPENAI_API_KEY_B", "YOUR_KEY_FOR_MODEL_B")
),
}
# =====================================
# 2) 로컬 파일 툴(폴더 목록/파일 읽기)
# =====================================
def list_files_in_folder(folder_path: str) -> List[str]:
"""지정 폴더 내 파일 목록 반환(파일만)."""
try:
items = os.listdir(folder_path)
except FileNotFoundError:
return []
files = [f for f in items if os.path.isfile(os.path.join(folder_path, f))]
return files
def read_text_file(file_path: str) -> str:
"""지정 경로의 텍스트 파일 내용 반환(UTF-8 가정)."""
if not os.path.exists(file_path):
return ""
try:
with open(file_path, "r", encoding="utf-8") as f:
return f.read()
except UnicodeDecodeError:
# 바이너리 등 텍스트가 아닌 경우
return ""
# 공통 툴 스키마(JSON Schema; Chat Completions/Responses에서 그대로 사용)
TOOLS = [
{
"type": "function",
"function": {
"name": "list_files_in_folder",
"description": "List file names (not directories) inside a given folder path.",
"parameters": {
"type": "object",
"properties": {
"folder_path": {
"type": "string",
"description": "Absolute or relative folder path."
}
},
"required": ["folder_path"],
"additionalProperties": False
}
}
},
{
"type": "function",
"function": {
"name": "read_text_file",
"description": "Read a UTF-8 text file and return its content as a string.",
"parameters": {
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "Absolute or relative path to the text file."
}
},
"required": ["file_path"],
"additionalProperties": False
}
}
}
]
# 툴 디스패처
def dispatch_tool(name: str, arguments: Dict[str, Any]) -> str:
if name == "list_files_in_folder":
folder = arguments.get("folder_path", "")
return json.dumps(list_files_in_folder(folder), ensure_ascii=False)
elif name == "read_text_file":
file_path = arguments.get("file_path", "")
return read_text_file(file_path)
else:
return f"Unknown tool: {name}"
# =====================================
# 3) Chat Completions로 함수 호출 테스트
# =====================================
def run_with_chat_completions(cfg: ModelConfig, user_task: str, tool_choice: str = "auto") -> str:
"""
Chat Completions API로 function calling 테스트.
- tools: TOOLS
- tool_choice: 'auto' 권장
- tool_calls -> 실제 파이썬 함수 실행 -> tool 메시지 첨부 -> 재호출
"""
client = OpenAI(base_url=cfg.base_url, api_key=cfg.api_key)
messages = [
{"role": "system", "content": "You are a helpful assistant that uses tools when needed."},
{"role": "user", "content": user_task},
]
# 1차 호출
resp = client.chat.completions.create(
model=cfg.name,
messages=messages,
tools=TOOLS,
tool_choice=tool_choice # 'auto' or {'type':'function','function':{'name':'...'}}
)
msg = resp.choices[0].message
# 함수 호출 루프
while getattr(msg, "tool_calls", None):
tool_msgs = []
for call in msg.tool_calls:
fn_name = call.function.name
fn_args = json.loads(call.function.arguments or "{}")
tool_output = dispatch_tool(fn_name, fn_args)
# tool 메시지 첨부
tool_msgs.append({
"role": "tool",
"tool_call_id": call.id,
"name": fn_name,
"content": tool_output
})
messages.append({"role": "assistant", "content": msg.content or "", "tool_calls": [tc.model_dump() for tc in msg.tool_calls]})
messages.extend(tool_msgs)
# 후속 호출(최종 답변 유도)
resp = client.chat.completions.create(
model=cfg.name,
messages=messages
)
msg = resp.choices[0].message
return msg.content or ""
# =====================================
# 4) Responses API로 함수 호출 테스트
# - 최신 SDK 규약에 맞춰 비스트리밍 루프 구성
# =====================================
def extract_text_from_responses_output(output_items: List[Any]) -> str:
"""Responses API의 output 배열에서 자연어 텍스트만 모아 반환."""
chunks = []
for item in output_items or []:
# item.type == "message" 인 경우, item.content[*] 중 type == "output_text"
if getattr(item, "type", "") == "message":
for c in getattr(item, "content", []) or []:
if getattr(c, "type", "") == "output_text":
chunks.append(getattr(c, "text", ""))
# 일부 모델은 최종에 바로 output_text만 내보내기도 함
if getattr(item, "type", "") == "output_text":
chunks.append(getattr(item, "text", ""))
return "\n".join([t for t in chunks if t])
def extract_tool_calls_from_responses_output(output_items: List[Any]) -> List[Dict[str, Any]]:
"""
Responses API의 output에서 tool_call 이벤트를 수집.
각 항목은 {id, name, arguments} 형태로 정규화해서 반환.
"""
calls = []
for item in output_items or []:
# 케이스1: item.type == "tool_call"
if getattr(item, "type", "") == "tool_call":
tc = {
"id": getattr(item, "id", None),
"name": getattr(item, "name", None),
"arguments": {}
}
# arguments는 문자열/딕셔너리 모두 가능성 고려
raw_args = getattr(item, "arguments", {})
if isinstance(raw_args, str):
try:
raw_args = json.loads(raw_args)
except Exception:
raw_args = {}
tc["arguments"] = raw_args or {}
calls.append(tc)
# 케이스2: item.type == "message" 내부 content[*]에 tool_calls 묶여있을 수 있음
if getattr(item, "type", "") == "message":
# 일부 SDK 버전에서 item.tool_calls 또는 item.content[*].tool_calls 형태
inner_calls = getattr(item, "tool_calls", None)
if inner_calls:
for c in inner_calls:
arg_obj = c.function.arguments if hasattr(c, "function") else {}
if isinstance(arg_obj, str):
try:
arg_obj = json.loads(arg_obj)
except Exception:
arg_obj = {}
calls.append({
"id": getattr(c, "id", None),
"name": getattr(c.function, "name", None) if hasattr(c, "function") else getattr(c, "name", None),
"arguments": arg_obj or {}
})
return calls
def run_with_responses(cfg: ModelConfig, user_task: str, tool_choice: str = "auto") -> str:
"""
Responses API로 function calling 테스트.
흐름:
1) responses.create(input=[{role:user, content:...}], tools=TOOLS, tool_choice=...)
2) tool_call이 있으면 로컬 함수 실행
3) responses.create(..., tool_outputs=[{tool_call_id, output}], response_id=prev.id) 로 이어서 호출
(최신 문서의 규약을 따름)
"""
client = OpenAI(base_url=cfg.base_url, api_key=cfg.api_key)
# 1차 호출
r = client.responses.create(
model=cfg.name,
input=[{"role": "user", "content": user_task}],
tools=TOOLS,
tool_choice=tool_choice
)
# 루프: 도구 호출이 끝날 때까지
final_text_parts: List[str] = []
guard = 0
while guard < 6: # 안전 상한
guard += 1
tool_calls = extract_tool_calls_from_responses_output(getattr(r, "output", []))
plain_text = extract_text_from_responses_output(getattr(r, "output", []))
if plain_text:
final_text_parts.append(plain_text)
if not tool_calls:
break
tool_outputs_payload = []
for tc in tool_calls:
out = dispatch_tool(tc["name"], tc["arguments"])
tool_outputs_payload.append({
"tool_call_id": tc["id"],
"output": out
})
# 이전 response와 이어 붙여 툴 결과를 전달
r = client.responses.create(
model=cfg.name,
tools=TOOLS,
response_id=r.id,
tool_outputs=tool_outputs_payload
)
return "\n".join([p for p in final_text_parts if p])
# =====================================
# 5) 실행부(4가지 조합)
# =====================================
DEFAULT_USER_TASK = (
"테스트: ./samples 폴더 안의 파일 목록을 보여주고, 텍스트 파일이 하나라도 있으면 "
"그 중 하나의 내용을 읽어서 핵심만 2~3문장으로 요약해줘. 필요하면 툴을 사용해."
)
def run_one(combo_api: str, model_key: str, user_task: str):
cfg = MODELS[model_key]
print(f"\n=== [{combo_api.upper()}] with Model {model_key} ({cfg.name}) ===")
if combo_api == "completions":
out = run_with_chat_completions(cfg, user_task)
elif combo_api == "responses":
out = run_with_responses(cfg, user_task)
else:
raise ValueError("combo_api must be 'completions' or 'responses'")
print("\n--- OUTPUT ---")
print(out or "(empty)")
def main():
parser = argparse.ArgumentParser(description="Test function-calling with OpenAI (4-way matrix).")
parser.add_argument("--api", choices=["completions", "responses", "all"], default="all",
help="Which API type to run.")
parser.add_argument("--model", choices=["A", "B", "both"], default="both",
help="Which model to use.")
parser.add_argument("--task", type=str, default=DEFAULT_USER_TASK, help="User task/prompt.")
args = parser.parse_args()
apis = ["completions", "responses"] if args.api == "all" else [args.api]
models = ["A", "B"] if args.model == "both" else [args.model]
# 샘플 폴더 안내
os.makedirs("./samples", exist_ok=True)
for api in apis:
for m in models:
run_one(api, m, args.task)
if __name__ == "__main__":
main()
사용 방법 간단 정리
- 환경 변수 세팅(권장)
-
Model A
-
OPENAI_API_KEY=…
-
OPENAI_BASE_URL=https://api.openai.com/v1
-
-
Model B
-
OPENAI_API_KEY_B=…
-
OPENAI_BASE_URL_B=https://api.openai.com/v1
-
- 테스트 파일 준비
- 프로젝트 루트에 samples 폴더를 만들고, 텍스트 파일을 1개 이상 넣는다. 예: samples/readme.txt
- 실행
-
네 조합 모두:
python test_openai_tools.py -
특정 API만:
python test_openai_tools.py –api completions
python test_openai_tools.py –api responses -
특정 모델만:
python test_openai_tools.py –model A
python test_openai_tools.py –model B -
프롬프트 커스터마이즈:
python test_openai_tools.py –task "여기에 원하는 테스트 지시문"
구현 포인트 설명(필요 시 점검할 체크리스트)
-
Chat Completions: response.choices[0].message.tool_calls를 확인하고, 각 툴을 실행한 뒤 tool 메시지를 추가해서 재호출하는 전형적 루프를 사용했어. 이것은 Function Calling 가이드의 권고 패턴이야. OpenAI 플랫폼
-
Responses API: responses.create(…) → output에서 tool_call 이벤트를 수집 → 실제 함수 실행 → tool_outputs와 이전 response_id를 함께 넘겨 후속 responses.create(…) 호출로 이어 붙이는 패턴이야. 이 흐름은 공식 Responses API 레퍼런스가 안내하는 방식과 맞물려 있어. OpenAI 플랫폼
-
tools 스키마는 두 API에서 동일하게 재사용.
-
로컬 파일 I/O는 안전하게 예외 처리했고, 텍스트 파일로 가정(UTF-8)했지만 디코딩 실패 시 빈 문자열을 반환해 모델이 적절히 안내하도록 했어.
문서 참고 링크
-
Function Calling 가이드(툴 정의/호출 흐름) OpenAI 플랫폼
-
Responses API 레퍼런스(응답 구조, tool_outputs 전달 방식) OpenAI 플랫폼
필요하면
-
모델명만 바꿔서(예: 최신 gpt-4.1 계열 등) 즉시 시험 가능
-
tool_choice를 "any"로 바꾸면 여러 툴 병렬 호출 경향을 더 시험할 수 있어
-
samples 폴더 경로나 읽을 파일 선택 로직을 더 엄격하게 지정하고 싶다면, 프롬프트를 고정(seed task)하거나 시스템 메시지쪽에 규칙을 더해도 좋아
원하는 부분 더 바꿔드릴게.
아니 내가 조사했을 때 responses용 툴 json이 있고, completion용이 따로 있어 조금 다르게 생겼던데 찾아서 반영해줘
좋은 지적이야. 실제로 “도구 정의(tools JSON)”는 두 API가 거의 같지만, Responses API는 실행·상태 전파 방식(response_id, tool_outputs 등)과 출력 구조가 다르고, 가이드에선 Responses 예시에서 엄격 파싱(strict) 옵션을 곁들여 쓰는 패턴이 자주 보여서 모양이 살짝 달라 보일 수 있어. 그래서 아래처럼 아예 두 벌의 스키마를 분리해 두고, Responses 쪽엔 엄격 파싱을 넣는 식으로 반영했어. 또한 Responses는 tool_outputs와 response_id를 통해 단계적으로 이어붙이는 점을 코드에 명확히 했어. 참고 문서: Function Calling 가이드와 Responses API 레퍼런스. OpenAI 플랫폼+2
아래 스크립트는 네 가지 조합(모델 A/B × completions/responses) 모두 함수 호출을 시험하고, 두 종류의 파일 툴을 제공해. 변경 포인트는 1) CHAT_TOOLS vs RESPONSES_TOOLS 분리, 2) Responses 루프에서 tool_outputs·response_id 사용을 더 엄격히, 3) Responses 출력 파서 보강이야.
# test_openai_tools_split_tools.py
# pip install --upgrade openai
import os, json, argparse
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
from openai import OpenAI
@dataclass
class ModelConfig:
name: str
base_url: str
api_key: str
MODELS = {
"A": ModelConfig(
name=os.getenv("MODEL_A_NAME", "gpt-4.1-mini"),
base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
api_key=os.getenv("OPENAI_API_KEY", "YOUR_KEY_FOR_MODEL_A"),
),
"B": ModelConfig(
name=os.getenv("MODEL_B_NAME", "gpt-4o-mini"),
base_url=os.getenv("OPENAI_BASE_URL_B", "https://api.openai.com/v1"),
api_key=os.getenv("OPENAI_API_KEY_B", "YOUR_KEY_FOR_MODEL_B"),
),
}
# 공통 실제 구현 함수
import os
def list_files_in_folder(folder_path: str) -> List[str]:
try:
return [f for f in os.listdir(folder_path) if os.path.isfile(os.path.join(folder_path, f))]
except FileNotFoundError:
return []
def read_text_file(file_path: str) -> str:
if not os.path.exists(file_path):
return ""
try:
with open(file_path, "r", encoding="utf-8") as f:
return f.read()
except UnicodeDecodeError:
return ""
def dispatch_tool(name: str, args: Dict[str, Any]) -> str:
if name == "list_files_in_folder":
return json.dumps(list_files_in_folder(args.get("folder_path", "")), ensure_ascii=False)
if name == "read_text_file":
return read_text_file(args.get("file_path", ""))
return f"Unknown tool: {name}"
# Chat Completions 전용 도구 JSON
# - 전통적인 function calling 스키마
CHAT_TOOLS = [
{
"type": "function",
"function": {
"name": "list_files_in_folder",
"description": "List file names (not directories) inside a given folder path.",
"parameters": {
"type": "object",
"properties": {
"folder_path": {"type": "string", "description": "Absolute or relative folder path."}
},
"required": ["folder_path"],
"additionalProperties": False
}
},
},
{
"type": "function",
"function": {
"name": "read_text_file",
"description": "Read a UTF-8 text file and return its content.",
"parameters": {
"type": "object",
"properties": {
"file_path": {"type": "string", "description": "Path to the text file."}
},
"required": ["file_path"],
"additionalProperties": False
}
},
},
]
# Responses 전용 도구 JSON
# - 스키마 필드는 유사하지만, Responses 예제 관례에 맞춰 strict를 켬
# (모델이 schema에 정확히 맞춰 인자 생성하도록 강제)
RESPONSES_TOOLS = [
{
"type": "function",
"function": {
"name": "list_files_in_folder",
"description": "List file names (not directories) inside a given folder path.",
"parameters": {
"type": "object",
"properties": {
"folder_path": {"type": "string", "description": "Absolute or relative folder path."}
},
"required": ["folder_path"],
"additionalProperties": False
},
"strict": True
},
},
{
"type": "function",
"function": {
"name": "read_text_file",
"description": "Read a UTF-8 text file and return its content.",
"parameters": {
"type": "object",
"properties": {
"file_path": {"type": "string", "description": "Path to the text file."}
},
"required": ["file_path"],
"additionalProperties": False
},
"strict": True
},
},
]
def run_with_chat_completions(cfg: ModelConfig, user_task: str, tools_choice: str = "auto") -> str:
client = OpenAI(base_url=cfg.base_url, api_key=cfg.api_key)
messages = [
{"role": "system", "content": "You are a helpful assistant that uses tools when needed."},
{"role": "user", "content": user_task},
]
resp = client.chat.completions.create(
model=cfg.name, messages=messages, tools=CHAT_TOOLS, tool_choice=tools_choice
)
msg = resp.choices[0].message
while getattr(msg, "tool_calls", None):
# 각 툴 실행
tool_msgs = []
for call in msg.tool_calls:
fn_name = call.function.name
fn_args = json.loads(call.function.arguments or "{}")
output = dispatch_tool(fn_name, fn_args)
tool_msgs.append({"role": "tool", "tool_call_id": call.id, "name": fn_name, "content": output})
# 대화 이어붙이기
messages.append({"role": "assistant", "content": msg.content or "", "tool_calls": [tc.model_dump() for tc in msg.tool_calls]})
messages.extend(tool_msgs)
resp = client.chat.completions.create(model=cfg.name, messages=messages)
msg = resp.choices[0].message
return msg.content or ""
# Responses 출력 파서
def _extract_text(output_items: List[Any]) -> str:
chunks: List[str] = []
for item in output_items or []:
if getattr(item, "type", "") == "message":
for c in getattr(item, "content", []) or []:
if getattr(c, "type", "") == "output_text":
chunks.append(getattr(c, "text", ""))
if getattr(item, "type", "") == "output_text":
chunks.append(getattr(item, "text", ""))
return "\n".join([t for t in chunks if t])
def _extract_tool_calls(output_items: List[Any]) -> List[Dict[str, Any]]:
calls: List[Dict[str, Any]] = []
for item in output_items or []:
if getattr(item, "type", "") == "tool_call":
raw_args = getattr(item, "arguments", {})
if isinstance(raw_args, str):
try: raw_args = json.loads(raw_args)
except Exception: raw_args = {}
calls.append({"id": getattr(item, "id", None), "name": getattr(item, "name", None), "arguments": raw_args or {}})
if getattr(item, "type", "") == "message":
inner_calls = getattr(item, "tool_calls", None)
if inner_calls:
for c in inner_calls:
raw_args = getattr(getattr(c, "function", {}), "arguments", {})
if isinstance(raw_args, str):
try: raw_args = json.loads(raw_args)
except Exception: raw_args = {}
calls.append({
"id": getattr(c, "id", None),
"name": getattr(c.function, "name", None) if hasattr(c, "function") else getattr(c, "name", None),
"arguments": raw_args or {}
})
return calls
def run_with_responses(cfg: ModelConfig, user_task: str, tools_choice: str = "auto") -> str:
client = OpenAI(base_url=cfg.base_url, api_key=cfg.api_key)
r = client.responses.create(
model=cfg.name,
input=[{"role": "user", "content": user_task}],
tools=RESPONSES_TOOLS,
tool_choice=tools_choice
)
final_text: List[str] = []
# Responses는 tool_outputs + response_id로 상태를 이어서 보낸다.
# 이어붙이기 패턴은 공식 레퍼런스의 흐름과 동일. :contentReference[oaicite:1]{index=1}
for _ in range(8): # 안전 상한
txt = _extract_text(getattr(r, "output", []))
if txt:
final_text.append(txt)
calls = _extract_tool_calls(getattr(r, "output", []))
if not calls:
break
tool_outputs = []
for tc in calls:
out = dispatch_tool(tc["name"], tc["arguments"])
tool_outputs.append({"tool_call_id": tc["id"], "output": out})
r = client.responses.create(
model=cfg.name,
tools=RESPONSES_TOOLS,
response_id=r.id, # 이전 응답과 연결
tool_outputs=tool_outputs # 툴 실행 결과 전달
)
return "\n\n".join([t for t in final_text if t]) or ""
DEFAULT_TASK = (
"테스트: ./samples 폴더의 파일 목록을 보여주고, 텍스트 파일이 있으면 하나 골라 내용을 읽고 2~3문장 요약. 필요하면 툴 사용."
)
def run_one(which_api: str, model_key: str, user_task: str):
cfg = MODELS[model_key]
print(f"\n=== [{which_api.upper()}] Model {model_key} ({cfg.name}) ===")
if which_api == "completions":
out = run_with_chat_completions(cfg, user_task)
elif which_api == "responses":
out = run_with_responses(cfg, user_task)
else:
raise ValueError("which_api must be 'completions' or 'responses'")
print("\n--- OUTPUT ---")
print(out or "(empty)")
if __name__ == "__main__":
os.makedirs("./samples", exist_ok=True)
parser = argparse.ArgumentParser()
parser.add_argument("--api", choices=["completions", "responses", "all"], default="all")
parser.add_argument("--model", choices=["A", "B", "both"], default="both")
parser.add_argument("--task", default=DEFAULT_TASK)
args = parser.parse_args()
apis = ["completions", "responses"] if args.api == "all" else [args.api]
models = ["A", "B"] if args.model == "both" else [args.model]
for a in apis:
for m in models:
run_one(a, m, args.task)
핵심 차이 정리 단계별
-
도구 정의 자체는 필드 구조가 동일(type=function, function.name/description/parameters)하되, Responses 쪽 예시에서는 strict를 켜서 JSON Schema를 엄격히 지키게 하는 패턴이 널리 쓰인다. 그래서 스키마가 “조금 다르게 생긴 것처럼” 보일 수 있다. OpenAI 플랫폼+1
-
Chat Completions는 messages 배열에 assistant.tool_calls가 나오고, 각 call.id에 대응하는 tool 메시지를 넣어 다시 chat.completions.create를 호출한다.
-
Responses는 output에 tool_call 항목이 나오며, 실제 도구 실행 후 tool_outputs 배열과 이전 응답의 response_id를 포함해 responses.create를 다시 호출해 상태를 이어 붙인다. 이게 가장 큰 차이점이다. OpenAI 플랫폼
이 버전으로 돌려보고, Responses에서 strict를 끄고 싶으면 RESPONSES_TOOLS에서 "strict": True만 빼면 돼. 모델·엔드포인트·키는 상단 MODELS에서 조정 가능하고, samples 폴더에 테스트 파일만 있으면 바로 네 조합을 모두 확인할 수 있어.
아니 예시를 좀 찾아봐.. from openai import OpenAI import json client = OpenAI() # 1. Define a list of callable tools for the model tools = [ { "type": "function", "name": "get_horoscope", "description": "Get today's horoscope for an astrological sign.", "parameters": { "type": "object", "properties": { "sign": { "type": "string", "description": "An astrological sign like Taurus or Aquarius", }, }, "required": ["sign"], }, }, ] def get_horoscope(sign): return f"{sign}: Next Tuesday you will befriend a baby otter." # Create a running input list we will add to over time input_list = [ {"role": "user", "content": "What is my horoscope? I am an Aquarius."} ] # 2. Prompt the model with tools defined response = client.responses.create( model="gpt-5", tools=tools, input=input_list, ) # Save function call outputs for subsequent requests input_list += response.output for item in response.output: if item.type == "function_call": if item.name == "get_horoscope": # 3. Execute the function logic for get_horoscope horoscope = get_horoscope(json.loads(item.arguments)) # 4. Provide function call results to the model input_list.append({ "type": "function_call_output", "call_id": item.call_id, "output": json.dumps({ "horoscope": horoscope }) }) print("Final input:") print(input_list) response = client.responses.create( model="gpt-5", instructions="Respond only with a horoscope generated by a tool.", tools=tools, input=input_list, ) # 5. The model should be able to give a response! print("Final output:") print(response.model_dump_json(indent=2)) print("\n" + response.output_text) 그러면 총 8개 조합이 나오지? 모델 x API 타입(completion, responses) x json 형태 이 8개 조합에 대해 각각 테스트하는 코드를 짜줘 이 때 실행 중 에러가 나는 경우에는 에러만 출력하고 나머지가 잘 수행되게 해줘.
좋아. “2개 모델 × 2개 API(completions, responses) × 2가지 tools JSON 형태(Completions 스타일 vs Responses 스타일)” = 총 8개 조합을 한 번에 돌려보는 테스트 스크립트를 준비했어. 각 조합은 개별 try/except로 감싸 에러가 나더라도 나머지가 계속 수행돼. tools JSON은 네가 준 Responses 스타일 예시(평평한 형태: type/name/parameters가 최상위)와, Chat Completions에서 흔히 쓰는 중첩 형태({"type":"function","function":{…}})를 각각 분리해 두었고, 일부 조합은 의도적으로 “맞지 않는 형태”를 넣었기 때문에 실패할 수도 있어(그 경우 에러만 출력). 참고로 Responses와 Chat Completions의 차이, strict 옵션 등은 최신 가이드에 정리돼 있어. OpenAI 플랫폼+2
아래 파일을 test_8combos_tools.py로 저장해서 실행하면 돼.
# test_8combos_tools.py
# Python 3.10+
# pip install --upgrade openai
import os
import json
import argparse
from dataclasses import dataclass
from typing import Any, Dict, List, Tuple, Optional
from openai import OpenAI
# =========================
# 0) 모델 설정
# =========================
@dataclass
class ModelConfig:
name: str
base_url: str
api_key: str
MODELS: Dict[str, ModelConfig] = {
"A": ModelConfig(
name=os.getenv("MODEL_A_NAME", "gpt-4.1-mini"),
base_url=os.getenv("OPENAI_BASE_URL_A", "https://api.openai.com/v1"),
api_key=os.getenv("OPENAI_API_KEY_A", "YOUR_KEY_FOR_MODEL_A"),
),
"B": ModelConfig(
name=os.getenv("MODEL_B_NAME", "gpt-4o-mini"),
base_url=os.getenv("OPENAI_BASE_URL_B", "https://api.openai.com/v1"),
api_key=os.getenv("OPENAI_API_KEY_B", "YOUR_KEY_FOR_MODEL_B"),
),
}
# =========================
# 1) 로컬 툴 구현
# =========================
def list_files_in_folder(folder_path: str) -> List[str]:
try:
return sorted([
f for f in os.listdir(folder_path)
if os.path.isfile(os.path.join(folder_path, f))
])
except Exception:
return []
def read_text_file(file_path: str) -> str:
if not os.path.exists(file_path):
return ""
try:
with open(file_path, "r", encoding="utf-8") as f:
return f.read()
except Exception:
return ""
def dispatch_tool(name: str, args: Dict[str, Any]) -> Dict[str, Any]:
"""
모든 툴의 결과를 dict로 통일해서 반환.
Responses의 function_call_output에 넣기 좋게 {"result": ...} 형태로 감싸준다.
"""
if name == "list_files_in_folder":
folder = args.get("folder_path", "")
return {"result": list_files_in_folder(folder)}
if name == "read_text_file":
path = args.get("file_path", "")
return {"result": read_text_file(path)}
return {"error": f"Unknown tool: {name}"}
# =========================
# 2) tools JSON 두 가지 형태
# =========================
# (A) Chat Completions 스타일: function 필드에 중첩
CHAT_STYLE_TOOLS = [
{
"type": "function",
"function": {
"name": "list_files_in_folder",
"description": "List file names (not directories) inside a given folder path.",
"parameters": {
"type": "object",
"properties": {
"folder_path": {"type": "string", "description": "Absolute or relative folder path."}
},
"required": ["folder_path"],
"additionalProperties": False,
},
},
},
{
"type": "function",
"function": {
"name": "read_text_file",
"description": "Read a UTF-8 text file and return its content.",
"parameters": {
"type": "object",
"properties": {
"file_path": {"type": "string", "description": "Absolute or relative file path."}
},
"required": ["file_path"],
"additionalProperties": False,
},
},
},
]
# (B) Responses 스타일: 네가 준 예시처럼 평평한 형태(일부 문서/예시에서 보이는 포맷)
RESPONSES_STYLE_TOOLS = [
{
"type": "function",
"name": "list_files_in_folder",
"description": "List file names (not directories) inside a given folder path.",
"parameters": {
"type": "object",
"properties": {
"folder_path": {"type": "string", "description": "Absolute or relative folder path."}
},
"required": ["folder_path"],
"additionalProperties": False,
},
# "strict": True # 필요 시 켜서 인자 스키마 준수 강제 (도움말 참조)
},
{
"type": "function",
"name": "read_text_file",
"description": "Read a UTF-8 text file and return its content.",
"parameters": {
"type": "object",
"properties": {
"file_path": {"type": "string", "description": "Absolute or relative file path."}
},
"required": ["file_path"],
"additionalProperties": False,
},
# "strict": True
},
]
# =========================
# 3) Chat Completions 쪽 함수 호출 루프
# =========================
def run_chat_completions(cfg: ModelConfig, user_task: str, tools_variant: str) -> str:
"""
tools_variant: "chat" 또는 "responses"
"chat" -> CHAT_STYLE_TOOLS 사용 (권장)
"responses" -> RESPONSES_STYLE_TOOLS 사용 (의도적 비호환 가능)
"""
client = OpenAI(base_url=cfg.base_url, api_key=cfg.api_key)
tools = CHAT_STYLE_TOOLS if tools_variant == "chat" else RESPONSES_STYLE_TOOLS
messages = [
{"role": "system", "content": "You are a helpful assistant that uses tools when needed."},
{"role": "user", "content": user_task},
]
resp = client.chat.completions.create(
model=cfg.name,
messages=messages,
tools=tools,
tool_choice="auto",
)
msg = resp.choices[0].message
# tool_calls 루프
guard = 0
while getattr(msg, "tool_calls", None) and guard < 8:
guard += 1
tool_msgs = []
for call in msg.tool_calls:
fn_name = call.function.name
# arguments는 문자열일 수 있음
try:
fn_args = json.loads(call.function.arguments or "{}")
except Exception:
fn_args = {}
tool_output = dispatch_tool(fn_name, fn_args)
tool_msgs.append({
"role": "tool",
"tool_call_id": call.id,
"name": fn_name,
"content": json.dumps(tool_output, ensure_ascii=False),
})
# 대화 이어붙이기
messages.append({
"role": "assistant",
"content": msg.content or "",
"tool_calls": [tc.model_dump() for tc in msg.tool_calls],
})
messages.extend(tool_msgs)
resp = client.chat.completions.create(model=cfg.name, messages=messages)
msg = resp.choices[0].message
return msg.content or ""
# =========================
# 4) Responses 쪽 함수 호출 루프
# =========================
def _collect_function_calls_from_output(output: List[Any]) -> List[Dict[str, Any]]:
"""
Responses API의 output 배열에서 function_call / tool_call 류 이벤트를 추출.
다양한 케이스를 최대한 관대하게 처리한다.
"""
calls: List[Dict[str, Any]] = []
for item in output or []:
itype = getattr(item, "type", None)
# 케이스 A: item.type == "function_call"
if itype == "function_call":
name = getattr(item, "name", None)
args = getattr(item, "arguments", {})
if isinstance(args, str):
try:
args = json.loads(args)
except Exception:
args = {}
calls.append({
"call_id": getattr(item, "call_id", None),
"name": name,
"arguments": args or {},
})
# 케이스 B: 일부 모델이 "tool_call" / item.message.tool_calls 형태로 줄 때
if itype == "message":
inner = getattr(item, "tool_calls", None)
if inner:
for c in inner:
nm = getattr(getattr(c, "function", {}), "name", None) if hasattr(c, "function") else getattr(c, "name", None)
args = getattr(getattr(c, "function", {}), "arguments", {})
if isinstance(args, str):
try:
args = json.loads(args)
except Exception:
args = {}
calls.append({
"call_id": getattr(c, "id", None),
"name": nm,
"arguments": args or {},
})
return calls
def _collect_text_from_output(output: List[Any]) -> str:
texts: List[str] = []
for item in output or []:
if getattr(item, "type", "") == "message":
for c in getattr(item, "content", []) or []:
if getattr(c, "type", "") == "output_text":
texts.append(getattr(c, "text", ""))
if getattr(item, "type", "") == "output_text":
texts.append(getattr(item, "text", ""))
return "\n".join([t for t in texts if t])
def run_responses(cfg: ModelConfig, user_task: str, tools_variant: str) -> str:
"""
tools_variant: "chat" 또는 "responses"
"responses" -> RESPONSES_STYLE_TOOLS 사용 (권장)
"chat" -> CHAT_STYLE_TOOLS 사용 (의도적 비호환 가능)
순수 Responses 방식: 네가 준 예시 패턴(input_list 축적 + function_call_output)로 구현.
"""
client = OpenAI(base_url=cfg.base_url, api_key=cfg.api_key)
tools = RESPONSES_STYLE_TOOLS if tools_variant == "responses" else CHAT_STYLE_TOOLS
# running input_list
input_list: List[Dict[str, Any]] = [
{"role": "user", "content": user_task}
]
r = client.responses.create(
model=cfg.name,
tools=tools,
input=input_list,
)
# function_call 처리
input_list += list(getattr(r, "output", []) or [])
# call들을 찾아 실행하고, function_call_output을 이어 붙임
calls = _collect_function_calls_from_output(getattr(r, "output", []))
for call in calls:
result = dispatch_tool(call["name"], call["arguments"])
input_list.append({
"type": "function_call_output",
"call_id": call["call_id"],
"output": json.dumps(result, ensure_ascii=False),
})
# 최종 응답 시도(지시문을 추가적으로 줄 수도 있음)
r2 = client.responses.create(
model=cfg.name,
instructions="Respond with a short summary of what you found.",
tools=tools,
input=input_list,
)
# 텍스트 추출(중간/최종)
final_text = []
txt1 = _collect_text_from_output(getattr(r, "output", []))
if txt1: final_text.append(txt1)
txt2 = _collect_text_from_output(getattr(r2, "output", []))
if txt2: final_text.append(txt2)
return "\n".join([t for t in final_text if t]) or ""
# =========================
# 5) 8개 조합 실행기
# =========================
DEFAULT_TASK = (
"테스트: ./samples 폴더의 파일 목록을 보여주고, 텍스트 파일이 있으면 하나 골라 내용을 읽고 2~3문장 요약. 필요하면 도구를 사용해."
)
def run_one(api_type: str, model_key: str, tools_variant: str, user_task: str) -> None:
cfg = MODELS[model_key]
header = f"[{api_type.upper()}] model={model_key}:{cfg.name} tools_json={tools_variant}"
print("\n" + "=" * (len(header) + 2))
print(f" {header} ")
print("=" * (len(header) + 2))
try:
if api_type == "completions":
out = run_chat_completions(cfg, user_task, tools_variant)
elif api_type == "responses":
out = run_responses(cfg, user_task, tools_variant)
else:
raise ValueError("api_type must be 'completions' or 'responses'")
print("\n--- OUTPUT ---")
print(out or "(empty)")
except Exception as e:
print("\n--- ERROR ---")
print(repr(e))
def main():
parser = argparse.ArgumentParser(description="Run 8 combinations: 2 models × 2 APIs × 2 tools JSON styles.")
parser.add_argument("--task", default=DEFAULT_TASK)
parser.add_argument("--models", choices=["A", "B", "both"], default="both")
parser.add_argument("--apis", choices=["completions", "responses", "both"], default="both")
# tools_variant를 "both"로 두면 각 API에서 chat/responses 두 가지 툴 JSON 모두 시도 → 8조합
parser.add_argument("--tools", choices=["chat", "responses", "both"], default="both")
args = parser.parse_args()
# 샘플 폴더 생성 안내
os.makedirs("./samples", exist_ok=True)
model_keys = ["A", "B"] if args.models == "both" else [args.models]
api_types = ["completions", "responses"] if args.apis == "both" else [args.apis]
tool_variants = ["chat", "responses"] if args.tools == "both" else [args.tools]
# 8조합: 2모델 × 2API × 2JSON
for mk in model_keys:
for api in api_types:
for tv in tool_variants:
run_one(api, mk, tv, args.task)
if __name__ == "__main__":
main()
실행 예시
python test_8combos_tools.py
python test_8combos_tools.py –apis completions –tools both
python test_8combos_tools.py –models A –apis responses –tools responses
포인트 요약
-
CHAT_STYLE_TOOLS와 RESPONSES_STYLE_TOOLS를 분리해 두었고, Responses 루프는 네가 붙여준 예시처럼 input_list에 function_call과 function_call_output을 누적하는 방식으로 구현했어. 일부 문서/예시에선 strict를 켜 JSON 스키마를 강제하니 필요하면 해당 키를 켜면 돼. OpenAI 플랫폼+1
-
일부 조합은 의도적으로 호환되지 않아 에러가 날 수 있어(예: Chat Completions에 Responses 스타일 tools). 모든 run은 try/except로 감싸 실패해도 나머지가 이어진다.
-
최신 차이점과 권장 흐름은 Responses vs Chat Completions 가이드에서 비교 가능해. OpenAI 플랫폼
더 엄격하게 Responses의 “response_id + tool_outputs” 방식으로 이어붙이는 버전도 원하면 따로 추가해줄게.
개선 필요 사항 1. responses api에서도 여러 번 콜을 하고 각 대화를 다 출력하고 싶어. 2. 입력 전체를 그냥 json으로 출력하고, 출력 전체도 그냥 json으로 다 저장해줘. 단, 콘솔 출력 색깔 지정을 통해서 (1) 입력 (2) 툴 요청 (3) 출력 이 구분되면 좋겠어. 3. 추가적으로 json formatting (structured json)도 추가해줘. structured json이 있는 버젼과 없는 버젼 각각 추가해주면 좋겠어. 총 16가지 조합이 될 거야. 각 api에 대한 structured formatting 예시는 아래 코드를 참고해 chat completions
from pydantic import BaseModel
from openai import OpenAI
client = OpenAI()
class CalendarEvent(BaseModel):
name: str
date: str
participants: list[str]
completion = client.chat.completions.parse(
model="gpt-4o-2024-08-06",
messages=[
{"role": "system", "content": "Extract the event information."},
{"role": "user", "content": "Alice and Bob are going to a science fair on Friday."},
],
response_format=CalendarEvent,
)
event = completion.choices[0].message.parsed
responses
from openai import OpenAI
from pydantic import BaseModel
client = OpenAI()
class CalendarEvent(BaseModel):
name: str
date: str
participants: list[str]
response = client.responses.parse(
model="gpt-4o-2024-08-06",
input=[
{"role": "system", "content": "Extract the event information."},
{
"role": "user",
"content": "Alice and Bob are going to a science fair on Friday.",
},
],
text_format=CalendarEvent,
)
event = response.output_parsed
좋아, 요구사항(2모델 × 2 API 타입 × 2 Tools JSON 스타일 × 2 Structured 포맷 유무 = 16조합)을 모두 만족하는 단일 실행 스크립트를 준비했어. 다음을 보장해.
-
Responses API에서 여러 번 호출(툴 호출 발생 시 루프)하며, 매 스텝의 대화 전개를 콘솔에 모두 출력.
-
각 조합마다 “입력 전체”와 “출력 전체”를 JSON으로 파일로 저장. 콘솔에선 색상으로 (입력 / 툴요청 / 출력) 구분.
-
Structured JSON 파싱 버전과 일반 버전 각각 제공.
-
에러는 조합별로 try/except로 고립 처리하여, 실패해도 나머지 조합 계속 실행.
아래 코드를 test_16combos.py로 저장해서 실행해. samples 폴더에 텍스트 파일 몇 개 두면 툴 동작을 쉽게 확인할 수 있어.
# test_16combos.py
# Python 3.10+
# pip install --upgrade openai pydantic
import os
import json
import time
import argparse
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple
from pydantic import BaseModel
from openai import OpenAI
# =========================
# ANSI 색상 유틸
# =========================
RESET = "\033[0m"
CYAN = "\033[36m" # 입력
YELLOW = "\033[33m" # 툴 요청/툴 출력
GREEN = "\033[32m" # 모델 출력
RED = "\033[31m" # 에러
# =========================
# 모델/엔드포인트/키 설정
# =========================
@dataclass
class ModelConfig:
name: str
base_url: str
api_key: str
MODELS: Dict[str, ModelConfig] = {
"A": ModelConfig(
name=os.getenv("MODEL_A_NAME", "gpt-4.1-mini"),
base_url=os.getenv("OPENAI_BASE_URL_A", "https://api.openai.com/v1"),
api_key=os.getenv("OPENAI_API_KEY_A", "YOUR_KEY_FOR_MODEL_A"),
),
"B": ModelConfig(
name=os.getenv("MODEL_B_NAME", "gpt-4o-mini"),
base_url=os.getenv("OPENAI_BASE_URL_B", "https://api.openai.com/v1"),
api_key=os.getenv("OPENAI_API_KEY_B", "YOUR_KEY_FOR_MODEL_B"),
),
}
# =========================
# 로컬 툴 구현
# =========================
def list_files_in_folder(folder_path: str) -> List[str]:
try:
return sorted([
f for f in os.listdir(folder_path)
if os.path.isfile(os.path.join(folder_path, f))
])
except Exception:
return []
def read_text_file(file_path: str) -> str:
if not os.path.exists(file_path):
return ""
try:
with open(file_path, "r", encoding="utf-8") as f:
return f.read()
except Exception:
return ""
def dispatch_tool(name: str, args: Dict[str, Any]) -> Dict[str, Any]:
if name == "list_files_in_folder":
return {"result": list_files_in_folder(args.get("folder_path", ""))}
if name == "read_text_file":
return {"result": read_text_file(args.get("file_path", ""))}
return {"error": f"Unknown tool: {name}"}
# =========================
# Tools JSON 2가지 스타일
# =========================
# Chat Completions 스타일 (function 필드 중첩)
CHAT_STYLE_TOOLS = [
{
"type": "function",
"function": {
"name": "list_files_in_folder",
"description": "List file names (not directories) inside a given folder path.",
"parameters": {
"type": "object",
"properties": {
"folder_path": {"type": "string", "description": "Absolute or relative folder path."}
},
"required": ["folder_path"],
"additionalProperties": False,
},
},
},
{
"type": "function",
"function": {
"name": "read_text_file",
"description": "Read a UTF-8 text file and return its content.",
"parameters": {
"type": "object",
"properties": {
"file_path": {"type": "string", "description": "Absolute or relative file path."}
},
"required": ["file_path"],
"additionalProperties": False,
},
},
},
]
# Responses 스타일 (평평한 형태; 일부 예시 포맷)
RESPONSES_STYLE_TOOLS = [
{
"type": "function",
"name": "list_files_in_folder",
"description": "List file names (not directories) inside a given folder path.",
"parameters": {
"type": "object",
"properties": {
"folder_path": {"type": "string", "description": "Absolute or relative folder path."}
},
"required": ["folder_path"],
"additionalProperties": False,
},
# "strict": True # 필요 시 활성화
},
{
"type": "function",
"name": "read_text_file",
"description": "Read a UTF-8 text file and return its content.",
"parameters": {
"type": "object",
"properties": {
"file_path": {"type": "string", "description": "Absolute or relative file path."}
},
"required": ["file_path"],
"additionalProperties": False,
},
# "strict": True
},
]
# =========================
# Structured JSON 포맷용 Pydantic
# =========================
class FileTaskResult(BaseModel):
files: List[str] = []
chosen_file: Optional[str] = None
summary: Optional[str] = None
# =========================
# 공통: 출력/저장 헬퍼
# =========================
def ensure_dirs(path: str):
os.makedirs(path, exist_ok=True)
def write_json(path: str, obj: Any):
with open(path, "w", encoding="utf-8") as f:
json.dump(obj, f, ensure_ascii=False, indent=2)
def label_for(model_key: str, api_type: str, tools_variant: str, structured: bool) -> str:
return f"{model_key}_{api_type}_{tools_variant}_{'structured' if structured else 'plain'}"
def print_section(title: str, color: str, payload: Any):
print(color + f"\n[{title}]" + RESET)
try:
print(json.dumps(payload, ensure_ascii=False, indent=2))
except Exception:
print(payload)
def now_id() -> str:
return time.strftime("%Y%m%d-%H%M%S")
# =========================
# Chat Completions 러너 (plain/structured)
# =========================
def run_chat(
cfg: ModelConfig,
tools_variant: str,
structured: bool,
user_task: str,
run_dir: str,
) -> None:
client = OpenAI(base_url=cfg.base_url, api_key=cfg.api_key)
tools = CHAT_STYLE_TOOLS if tools_variant == "chat" else RESPONSES_STYLE_TOOLS
# 대화 시작
messages: List[Dict[str, Any]] = [
{"role": "system", "content": "You are a helpful assistant that uses tools when needed."},
{"role": "user", "content": user_task},
]
# 입력/출력 전체 저장을 위한 구조
input_snapshot: Dict[str, Any] = {"messages": messages.copy(), "tools": tools}
responses_dump: List[Dict[str, Any]] = []
print_section("입력(Chat Completions)", CYAN, input_snapshot)
if structured:
# structured: parse() 사용
completion = client.chat.completions.parse(
model=cfg.name,
messages=messages,
tools=tools,
response_format=FileTaskResult,
tool_choice="auto",
)
responses_dump.append(json.loads(completion.model_dump_json()))
msg = completion.choices[0].message
# function calling 루프 (parse 결과에서도 tool_calls 접근 가능하도록 시도)
guard = 0
while getattr(msg, "tool_calls", None) and guard < 12:
guard += 1
tool_msgs = []
for call in msg.tool_calls:
fn_name = getattr(call.function, "name", None)
try:
fn_args = json.loads(call.function.arguments or "{}")
except Exception:
fn_args = {}
print_section("툴 요청(Chat Completions)", YELLOW, {"name": fn_name, "arguments": fn_args})
tool_output = dispatch_tool(fn_name, fn_args)
tool_msgs.append({
"role": "tool",
"tool_call_id": call.id,
"name": fn_name,
"content": json.dumps(tool_output, ensure_ascii=False),
})
messages.append({"role": "assistant", "content": msg.content or "", "tool_calls": [tc.model_dump() for tc in msg.tool_calls]})
messages.extend(tool_msgs)
completion = client.chat.completions.parse(
model=cfg.name,
messages=messages,
response_format=FileTaskResult,
)
responses_dump.append(json.loads(completion.model_dump_json()))
msg = completion.choices[0].message
# 최종 구조화 결과
parsed: Optional[FileTaskResult] = getattr(msg, "parsed", None)
out_to_print: Dict[str, Any] = {"text": msg.content, "parsed": parsed.model_dump() if parsed else None}
print_section("출력(Chat Completions, structured)", GREEN, out_to_print)
else:
# plain: create() 사용
resp = client.chat.completions.create(
model=cfg.name, messages=messages, tools=tools, tool_choice="auto"
)
responses_dump.append(json.loads(resp.model_dump_json()))
msg = resp.choices[0].message
guard = 0
while getattr(msg, "tool_calls", None) and guard < 12:
guard += 1
tool_msgs = []
for call in msg.tool_calls:
fn_name = call.function.name
try:
fn_args = json.loads(call.function.arguments or "{}")
except Exception:
fn_args = {}
print_section("툴 요청(Chat Completions)", YELLOW, {"name": fn_name, "arguments": fn_args})
tool_output = dispatch_tool(fn_name, fn_args)
tool_msgs.append({
"role": "tool",
"tool_call_id": call.id,
"name": fn_name,
"content": json.dumps(tool_output, ensure_ascii=False),
})
messages.append({"role": "assistant", "content": msg.content or "", "tool_calls": [tc.model_dump() for tc in msg.tool_calls]})
messages.extend(tool_msgs)
resp = client.chat.completions.create(model=cfg.name, messages=messages)
responses_dump.append(json.loads(resp.model_dump_json()))
msg = resp.choices[0].message
print_section("출력(Chat Completions, plain)", GREEN, {"text": msg.content})
# 파일 저장
write_json(os.path.join(run_dir, "input.json"), input_snapshot)
write_json(os.path.join(run_dir, "outputs.json"), responses_dump)
# =========================
# Responses 러너 (plain/structured)
# - 여러 번 호출하며 매 스텝 출력
# =========================
def _collect_function_calls_from_output(output: List[Any]) -> List[Dict[str, Any]]:
calls: List[Dict[str, Any]] = []
for item in output or []:
itype = getattr(item, "type", None)
if itype == "function_call":
name = getattr(item, "name", None)
args = getattr(item, "arguments", {})
if isinstance(args, str):
try:
args = json.loads(args)
except Exception:
args = {}
calls.append({"call_id": getattr(item, "call_id", None), "name": name, "arguments": args or {}})
if itype == "message":
inner = getattr(item, "tool_calls", None)
if inner:
for c in inner:
nm = getattr(getattr(c, "function", {}), "name", None) if hasattr(c, "function") else getattr(c, "name", None)
args = getattr(getattr(c, "function", {}), "arguments", {})
if isinstance(args, str):
try:
args = json.loads(args)
except Exception:
args = {}
calls.append({"call_id": getattr(c, "id", None), "name": nm, "arguments": args or {}})
return calls
def _collect_text_from_output(output: List[Any]) -> str:
texts: List[str] = []
for item in output or []:
if getattr(item, "type", "") == "message":
for c in getattr(item, "content", []) or []:
if getattr(c, "type", "") == "output_text":
texts.append(getattr(c, "text", ""))
if getattr(item, "type", "") == "output_text":
texts.append(getattr(item, "text", ""))
return "\n".join([t for t in texts if t])
def run_responses(
cfg: ModelConfig,
tools_variant: str,
structured: bool,
user_task: str,
run_dir: str,
) -> None:
client = OpenAI(base_url=cfg.base_url, api_key=cfg.api_key)
tools = RESPONSES_STYLE_TOOLS if tools_variant == "responses" else CHAT_STYLE_TOOLS
input_list: List[Dict[str, Any]] = [
{"role": "system", "content": "You are a helpful assistant that uses tools when needed."},
{"role": "user", "content": user_task},
]
input_snapshot: Dict[str, Any] = {"input": input_list.copy(), "tools": tools}
print_section("입력(Responses)", CYAN, input_snapshot)
responses_dump: List[Dict[str, Any]] = []
if structured:
# structured: responses.parse(text_format=...)
r = client.responses.parse(
model=cfg.name,
input=input_list,
tools=tools,
text_format=FileTaskResult,
)
responses_dump.append(json.loads(r.model_dump_json()))
print_section("출력(Responses step 1)", GREEN, json.loads(r.model_dump_json()))
input_list += list(getattr(r, "output", []) or [])
# 툴 호출 탐지 → 실행 → function_call_output 추가 → 이어서 또 parse
guard = 0
while True and guard < 12:
guard += 1
calls = _collect_function_calls_from_output(getattr(r, "output", []))
if not calls:
break
for call in calls:
print_section("툴 요청(Responses)", YELLOW, {"name": call["name"], "arguments": call["arguments"]})
result = dispatch_tool(call["name"], call["arguments"])
input_list.append({
"type": "function_call_output",
"call_id": call["call_id"],
"output": json.dumps(result, ensure_ascii=False),
})
r = client.responses.parse(
model=cfg.name,
input=input_list,
tools=tools,
text_format=FileTaskResult,
)
responses_dump.append(json.loads(r.model_dump_json()))
print_section("출력(Responses next)", GREEN, json.loads(r.model_dump_json()))
input_list += list(getattr(r, "output", []) or [])
# 최종 파싱 결과
parsed = getattr(r, "output_parsed", None)
print_section("출력 요약(Responses structured)", GREEN, parsed.model_dump() if parsed else None)
else:
# plain: responses.create()
r = client.responses.create(
model=cfg.name,
input=input_list,
tools=tools,
tool_choice="auto",
)
responses_dump.append(json.loads(r.model_dump_json()))
print_section("출력(Responses step 1)", GREEN, json.loads(r.model_dump_json()))
input_list += list(getattr(r, "output", []) or [])
guard = 0
while True and guard < 12:
guard += 1
calls = _collect_function_calls_from_output(getattr(r, "output", []))
if not calls:
break
tool_outputs = []
for call in calls:
print_section("툴 요청(Responses)", YELLOW, {"name": call["name"], "arguments": call["arguments"]})
result = dispatch_tool(call["name"], call["arguments"])
# function_call_output을 input에 누적
input_list.append({
"type": "function_call_output",
"call_id": call["call_id"],
"output": json.dumps(result, ensure_ascii=False),
})
# 다음 턴
r = client.responses.create(
model=cfg.name,
input=input_list,
tools=tools,
)
responses_dump.append(json.loads(r.model_dump_json()))
print_section("출력(Responses next)", GREEN, json.loads(r.model_dump_json()))
input_list += list(getattr(r, "output", []) or [])
final_text = _collect_text_from_output(getattr(r, "output", []))
print_section("출력 요약(Responses plain)", GREEN, {"text": final_text})
# 파일 저장
write_json(os.path.join(run_dir, "input.json"), input_snapshot)
write_json(os.path.join(run_dir, "outputs.json"), responses_dump)
# =========================
# 16개 조합 실행기
# 2 모델 × 2 API × 2 Tools JSON × 2 Structured
# =========================
DEFAULT_TASK = (
"테스트: ./samples 폴더의 파일 목록을 보여주고, 텍스트 파일이 있으면 하나를 골라 내용을 읽고 2~3문장으로 요약해줘. 필요하면 툴을 사용해."
)
def run_one(
model_key: str,
api_type: str,
tools_variant: str,
structured: bool,
user_task: str,
base_runs_dir: str,
):
cfg = MODELS[model_key]
combo_label = label_for(model_key, api_type, tools_variant, structured)
run_dir = os.path.join(base_runs_dir, combo_label)
ensure_dirs(run_dir)
header = f"[{api_type.upper()}] model={model_key}:{cfg.name} tools={tools_variant} structured={'on' if structured else 'off'}"
print("\n" + "=" * (len(header) + 2))
print(f" {header} ")
print("=" * (len(header) + 2))
try:
if api_type == "completions":
run_chat(cfg, tools_variant, structured, user_task, run_dir)
elif api_type == "responses":
run_responses(cfg, tools_variant, structured, user_task, run_dir)
else:
raise ValueError("api_type must be 'completions' or 'responses'")
except Exception as e:
print(RED + "\n--- ERROR ---" + RESET)
print(repr(e))
# 에러만 출력하고 계속
def main():
parser = argparse.ArgumentParser(description="Run 16 combos: 2 models × 2 APIs × 2 Tools JSON × 2 Structured.")
parser.add_argument("--task", default=DEFAULT_TASK)
parser.add_argument("--models", choices=["A", "B", "both"], default="both")
parser.add_argument("--apis", choices=["completions", "responses", "both"], default="both")
parser.add_argument("--tools", choices=["chat", "responses", "both"], default="both")
parser.add_argument("--structured", choices=["on", "off", "both"], default="both")
parser.add_argument("--runs_dir", default=f"./runs/{now_id()}")
args = parser.parse_args()
os.makedirs("./samples", exist_ok=True)
ensure_dirs(args.runs_dir)
model_keys = ["A", "B"] if args.models == "both" else [args.models]
api_types = ["completions", "responses"] if args.apis == "both" else [args.apis]
tool_variants = ["chat", "responses"] if args.tools == "both" else [args.tools]
structured_opts = [True, False] if args.structured == "both" else [args.structured == "on"]
for mk in model_keys:
for api in api_types:
for tv in tool_variants:
for st in structured_opts:
run_one(mk, api, tv, st, args.task, args.runs_dir)
if __name__ == "__main__":
main()
사용법 요약
- 환경 변수 설정(예시)
-
모델 A:
OPENAI_API_KEY_A, OPENAI_BASE_URL_A=https://api.openai.com/v1, MODEL_A_NAME -
모델 B:
OPENAI_API_KEY_B, OPENAI_BASE_URL_B=https://api.openai.com/v1, MODEL_B_NAME
-
샘플 준비
프로젝트 루트에 samples 폴더를 만들고 텍스트 파일을 1개 이상 넣어둔다. -
실행 예시
-
16개 전체 조합 실행
python test_16combos.py -
Responses만, structured/plain 둘 다, tools는 responses 스타일만
python test_16combos.py –apis responses –tools responses –structured both -
모델 A만, Chat Completions만, structured만
python test_16combos.py –models A –apis completions –structured on
동작 정리
-
콘솔 색상으로 입력(청록), 툴 요청(노랑), 출력(초록)을 구분.
-
각 조합은 runs/타임스탬프/모델_API_tools_structured/ 아래에 input.json, outputs.json 저장.
-
Responses API는 툴 호출이 계속 나오면 루프를 반복하고 매 스텝 출력 및 누적 저장.
-
Structured 모드에선 chat.completions.parse / responses.parse를 사용해 Pydantic 모델(FileTaskResult)로 결과를 파싱해 별도 요약도 콘솔에 표시.
원하는 추가 포맷(예: 파일명 규칙, 더 자세한 로깅, strict 옵션 상시 on 등)이 있으면 알려줘.
출력 json 포맷은 task에서 요청한 걸 대답할 수 있게 해주면 좋을 것 같아. 예를 들어 filename: 파일명 contents: 내용
좋아! “결과 JSON 포맷을 task에 바로 쓰이게(filename, contents)”로 맞춘 버전을 아래처럼 업데이트했어. 핵심은 structured 모드에서 파싱 모델을 {"filename": str, "contents": str}로 고정하고, 모델에게 “contents에는 요약을 넣어라”라고 지시하는 것. plain 모드에서는 기존 그대로 자유 텍스트를 출력하고 전체 원시 JSON 로그만 저장하며, structured 모드에서는 추가로 result.json에 {"filename","contents"}를 저장해.
아래 스크립트는 이전 16조합 실행기(test_16combos.py)를 대체하는 완전판이야. 바뀐 점만 빠르게 요약하고 바로 코드 줄게.
변경 요약
- Structured 파싱 모델 교체
-
Pydantic 모델을
TaskAnswer로 정의하고 필드를 filename, contents 두 개로 통일. -
contents는 “읽은 파일의 2~3문장 요약”을 넣도록 system/instructions에 명확히 지시.
- Structured 모드에서 결과 JSON 저장
-
각 조합 폴더에
result.json으로{"filename","contents"}저장. -
기존처럼
input.json,outputs.json도 계속 저장.
- 콘솔 색상 출력은 동일하게 유지
- 입력(CYAN), 툴 요청(YELLOW), 출력(GREEN)으로 구분.
- Responses API는 다단 호출 유지
- 각 step의 출력과 툴 요청을 모두 콘솔에 표시하고, outputs.json에 누적 저장.
코드
# test_16combos_v2.py
# Python 3.10+
# pip install --upgrade openai pydantic
import os
import json
import time
import argparse
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
from pydantic import BaseModel
from openai import OpenAI
# ===== ANSI 색상 =====
RESET = "\033[0m"
CYAN = "\033[36m" # 입력
YELLOW = "\033[33m" # 툴 요청/툴 출력
GREEN = "\033[32m" # 모델 출력
RED = "\033[31m" # 에러
# ===== 모델 설정 =====
@dataclass
class ModelConfig:
name: str
base_url: str
api_key: str
MODELS: Dict[str, ModelConfig] = {
"A": ModelConfig(
name=os.getenv("MODEL_A_NAME", "gpt-4.1-mini"),
base_url=os.getenv("OPENAI_BASE_URL_A", "https://api.openai.com/v1"),
api_key=os.getenv("OPENAI_API_KEY_A", "YOUR_KEY_FOR_MODEL_A"),
),
"B": ModelConfig(
name=os.getenv("MODEL_B_NAME", "gpt-4o-mini"),
base_url=os.getenv("OPENAI_BASE_URL_B", "https://api.openai.com/v1"),
api_key=os.getenv("OPENAI_API_KEY_B", "YOUR_KEY_FOR_MODEL_B"),
),
}
# ===== 로컬 툴 구현 =====
def list_files_in_folder(folder_path: str) -> List[str]:
try:
return sorted([f for f in os.listdir(folder_path) if os.path.isfile(os.path.join(folder_path, f))])
except Exception:
return []
def read_text_file(file_path: str) -> str:
if not os.path.exists(file_path):
return ""
try:
with open(file_path, "r", encoding="utf-8") as f:
return f.read()
except Exception:
return ""
def dispatch_tool(name: str, args: Dict[str, Any]) -> Dict[str, Any]:
if name == "list_files_in_folder":
return {"result": list_files_in_folder(args.get("folder_path", ""))}
if name == "read_text_file":
return {"result": read_text_file(args.get("file_path", ""))}
return {"error": f"Unknown tool: {name}"}
# ===== Tools JSON 2가지 =====
CHAT_STYLE_TOOLS = [
{
"type": "function",
"function": {
"name": "list_files_in_folder",
"description": "List file names (not directories) inside a given folder path.",
"parameters": {
"type": "object",
"properties": {"folder_path": {"type": "string", "description": "Absolute or relative folder path."}},
"required": ["folder_path"],
"additionalProperties": False,
},
},
},
{
"type": "function",
"function": {
"name": "read_text_file",
"description": "Read a UTF-8 text file and return its content.",
"parameters": {
"type": "object",
"properties": {"file_path": {"type": "string", "description": "Absolute or relative file path."}},
"required": ["file_path"],
"additionalProperties": False,
},
},
},
]
RESPONSES_STYLE_TOOLS = [
{
"type": "function",
"name": "list_files_in_folder",
"description": "List file names (not directories) inside a given folder path.",
"parameters": {
"type": "object",
"properties": {"folder_path": {"type": "string", "description": "Absolute or relative folder path."}},
"required": ["folder_path"],
"additionalProperties": False,
},
},
{
"type": "function",
"name": "read_text_file",
"description": "Read a UTF-8 text file and return its content.",
"parameters": {
"type": "object",
"properties": {"file_path": {"type": "string", "description": "Absolute or relative file path."}},
"required": ["file_path"],
"additionalProperties": False,
},
},
]
# ===== Structured JSON 스키마 =====
class TaskAnswer(BaseModel):
filename: str
contents: str # 선택된 파일의 2~3문장 요약(모델이 생성)
# ===== 유틸 =====
def ensure_dirs(path: str): os.makedirs(path, exist_ok=True)
def write_json(path: str, obj: Any):
with open(path, "w", encoding="utf-8") as f:
json.dump(obj, f, ensure_ascii=False, indent=2)
def now_id() -> str: return time.strftime("%Y%m%d-%H%M%S")
def print_section(title: str, color: str, payload: Any):
print(color + f"\n[{title}]" + RESET)
try:
print(json.dumps(payload, ensure_ascii=False, indent=2))
except Exception:
print(payload)
def combo_label(model_key: str, api_type: str, tools_variant: str, structured: bool) -> str:
return f"{model_key}_{api_type}_{tools_variant}_{'structured' if structured else 'plain'}"
# ===== Chat Completions 러너 =====
def run_chat(cfg: ModelConfig, tools_variant: str, structured: bool, user_task: str, run_dir: str) -> None:
client = OpenAI(base_url=cfg.base_url, api_key=cfg.api_key)
tools = CHAT_STYLE_TOOLS if tools_variant == "chat" else RESPONSES_STYLE_TOOLS
system_msg = (
"You are a helpful assistant that uses tools when needed. "
"Task: Show files under ./samples, pick one text file, read it, and provide a 2-3 sentence summary. "
"If using structured output, respond ONLY with JSON having keys: filename, contents."
)
messages = [
{"role": "system", "content": system_msg},
{"role": "user", "content": user_task},
]
input_snapshot = {"messages": messages.copy(), "tools": tools}
responses_dump: List[Dict[str, Any]] = []
print_section("입력(Chat Completions)", CYAN, input_snapshot)
if structured:
completion = client.chat.completions.parse(
model=cfg.name,
messages=messages,
tools=tools,
response_format=TaskAnswer,
tool_choice="auto",
)
responses_dump.append(json.loads(completion.model_dump_json()))
msg = completion.choices[0].message
guard = 0
while getattr(msg, "tool_calls", None) and guard < 12:
guard += 1
tool_msgs = []
for call in msg.tool_calls:
fn_name = call.function.name
try:
fn_args = json.loads(call.function.arguments or "{}")
except Exception:
fn_args = {}
print_section("툴 요청(Chat)", YELLOW, {"name": fn_name, "arguments": fn_args})
tool_output = dispatch_tool(fn_name, fn_args)
tool_msgs.append({"role": "tool", "tool_call_id": call.id, "name": fn_name,
"content": json.dumps(tool_output, ensure_ascii=False)})
messages.append({"role": "assistant", "content": msg.content or "",
"tool_calls": [tc.model_dump() for tc in msg.tool_calls]})
messages.extend(tool_msgs)
completion = client.chat.completions.parse(
model=cfg.name,
messages=messages,
response_format=TaskAnswer,
)
responses_dump.append(json.loads(completion.model_dump_json()))
msg = completion.choices[0].message
parsed: Optional[TaskAnswer] = getattr(msg, "parsed", None)
out_obj = parsed.model_dump() if parsed else {"filename": "", "contents": msg.content or ""}
print_section("출력(Chat structured)", GREEN, out_obj)
write_json(os.path.join(run_dir, "result.json"), out_obj)
else:
resp = client.chat.completions.create(
model=cfg.name, messages=messages, tools=tools, tool_choice="auto"
)
responses_dump.append(json.loads(resp.model_dump_json()))
msg = resp.choices[0].message
guard = 0
while getattr(msg, "tool_calls", None) and guard < 12:
guard += 1
tool_msgs = []
for call in msg.tool_calls:
fn_name = call.function.name
try:
fn_args = json.loads(call.function.arguments or "{}")
except Exception:
fn_args = {}
print_section("툴 요청(Chat)", YELLOW, {"name": fn_name, "arguments": fn_args})
tool_output = dispatch_tool(fn_name, fn_args)
tool_msgs.append({"role": "tool", "tool_call_id": call.id, "name": fn_name,
"content": json.dumps(tool_output, ensure_ascii=False)})
messages.append({"role": "assistant", "content": msg.content or "",
"tool_calls": [tc.model_dump() for tc in msg.tool_calls]})
messages.extend(tool_msgs)
resp = client.chat.completions.create(model=cfg.name, messages=messages)
responses_dump.append(json.loads(resp.model_dump_json()))
msg = resp.choices[0].message
print_section("출력(Chat plain)", GREEN, {"text": msg.content})
write_json(os.path.join(run_dir, "input.json"), input_snapshot)
write_json(os.path.join(run_dir, "outputs.json"), responses_dump)
# ===== Responses 러너 =====
def _collect_function_calls(output: List[Any]) -> List[Dict[str, Any]]:
calls: List[Dict[str, Any]] = []
for item in output or []:
itype = getattr(item, "type", None)
if itype == "function_call":
args = getattr(item, "arguments", {})
if isinstance(args, str):
try: args = json.loads(args)
except Exception: args = {}
calls.append({"call_id": getattr(item, "call_id", None), "name": getattr(item, "name", None), "arguments": args or {}})
if itype == "message":
inner = getattr(item, "tool_calls", None)
if inner:
for c in inner:
nm = getattr(getattr(c, "function", {}), "name", None) if hasattr(c, "function") else getattr(c, "name", None)
args = getattr(getattr(c, "function", {}), "arguments", {})
if isinstance(args, str):
try: args = json.loads(args)
except Exception: args = {}
calls.append({"call_id": getattr(c, "id", None), "name": nm, "arguments": args or {}})
return calls
def _collect_text(output: List[Any]) -> str:
texts: List[str] = []
for item in output or []:
if getattr(item, "type", "") == "message":
for c in getattr(item, "content", []) or []:
if getattr(c, "type", "") == "output_text":
texts.append(getattr(c, "text", ""))
if getattr(item, "type", "") == "output_text":
texts.append(getattr(item, "text", ""))
return "\n".join([t for t in texts if t])
def run_responses(cfg: ModelConfig, tools_variant: str, structured: bool, user_task: str, run_dir: str) -> None:
client = OpenAI(base_url=cfg.base_url, api_key=cfg.api_key)
tools = RESPONSES_STYLE_TOOLS if tools_variant == "responses" else CHAT_STYLE_TOOLS
system_inst = (
"You are a helpful assistant that uses tools when needed. "
"Task: Show files under ./samples, pick one text file, read it, and provide a 2-3 sentence summary. "
"If structured parsing is enabled, respond ONLY with JSON having keys: filename, contents."
)
input_list: List[Dict[str, Any]] = [
{"role": "system", "content": system_inst},
{"role": "user", "content": user_task},
]
input_snapshot = {"input": input_list.copy(), "tools": tools}
print_section("입력(Responses)", CYAN, input_snapshot)
responses_dump: List[Dict[str, Any]] = []
if structured:
r = client.responses.parse(
model=cfg.name,
input=input_list,
tools=tools,
text_format=TaskAnswer,
)
responses_dump.append(json.loads(r.model_dump_json()))
print_section("출력(Responses step 1)", GREEN, json.loads(r.model_dump_json()))
input_list += list(getattr(r, "output", []) or [])
guard = 0
while guard < 12:
guard += 1
calls = _collect_function_calls(getattr(r, "output", []))
if not calls:
break
for call in calls:
print_section("툴 요청(Responses)", YELLOW, {"name": call["name"], "arguments": call["arguments"]})
result = dispatch_tool(call["name"], call["arguments"])
input_list.append({
"type": "function_call_output",
"call_id": call["call_id"],
"output": json.dumps(result, ensure_ascii=False),
})
r = client.responses.parse(model=cfg.name, input=input_list, tools=tools, text_format=TaskAnswer)
responses_dump.append(json.loads(r.model_dump_json()))
print_section("출력(Responses next)", GREEN, json.loads(r.model_dump_json()))
input_list += list(getattr(r, "output", []) or [])
parsed = getattr(r, "output_parsed", None)
out_obj = parsed.model_dump() if parsed else {"filename": "", "contents": _collect_text(getattr(r, "output", []))}
print_section("출력 요약(Responses structured)", GREEN, out_obj)
write_json(os.path.join(run_dir, "result.json"), out_obj)
else:
r = client.responses.create(model=cfg.name, input=input_list, tools=tools, tool_choice="auto")
responses_dump.append(json.loads(r.model_dump_json()))
print_section("출력(Responses step 1)", GREEN, json.loads(r.model_dump_json()))
input_list += list(getattr(r, "output", []) or [])
guard = 0
while guard < 12:
guard += 1
calls = _collect_function_calls(getattr(r, "output", []))
if not calls:
break
for call in calls:
print_section("툴 요청(Responses)", YELLOW, {"name": call["name"], "arguments": call["arguments"]})
result = dispatch_tool(call["name"], call["arguments"])
input_list.append({
"type": "function_call_output",
"call_id": call["call_id"],
"output": json.dumps(result, ensure_ascii=False),
})
r = client.responses.create(model=cfg.name, input=input_list, tools=tools)
responses_dump.append(json.loads(r.model_dump_json()))
print_section("출력(Responses next)", GREEN, json.loads(r.model_dump_json()))
input_list += list(getattr(r, "output", []) or [])
final_text = _collect_text(getattr(r, "output", []))
print_section("출력 요약(Responses plain)", GREEN, {"text": final_text})
write_json(os.path.join(run_dir, "input.json"), input_snapshot)
write_json(os.path.join(run_dir, "outputs.json"), responses_dump)
# ===== 16개 조합 실행기 =====
DEFAULT_TASK = (
"테스트: ./samples 폴더의 파일 목록을 보여주고, 텍스트 파일이 있으면 하나를 골라 내용을 읽고 2~3문장으로 요약해줘. 필요하면 툴을 사용해."
)
def run_one(model_key: str, api_type: str, tools_variant: str, structured: bool, user_task: str, base_runs_dir: str):
cfg = MODELS[model_key]
label = combo_label(model_key, api_type, tools_variant, structured)
run_dir = os.path.join(base_runs_dir, label)
ensure_dirs(run_dir)
header = f"[{api_type.upper()}] model={model_key}:{cfg.name} tools={tools_variant} structured={'on' if structured else 'off'}"
print("\n" + "=" * (len(header) + 2))
print(f" {header} ")
print("=" * (len(header) + 2))
try:
if api_type == "completions":
run_chat(cfg, tools_variant, structured, user_task, run_dir)
elif api_type == "responses":
run_responses(cfg, tools_variant, structured, user_task, run_dir)
else:
raise ValueError("api_type must be 'completions' or 'responses'")
except Exception as e:
print(RED + "\n--- ERROR ---" + RESET)
print(repr(e)) # 에러만 출력하고 계속
def main():
parser = argparse.ArgumentParser(description="Run 16 combos with JSON result format {filename, contents}.")
parser.add_argument("--task", default=DEFAULT_TASK)
parser.add_argument("--models", choices=["A", "B", "both"], default="both")
parser.add_argument("--apis", choices=["completions", "responses", "both"], default="both")
parser.add_argument("--tools", choices=["chat", "responses", "both"], default="both")
parser.add_argument("--structured", choices=["on", "off", "both"], default="both")
parser.add_argument("--runs_dir", default=f"./runs/{now_id()}")
args = parser.parse_args()
os.makedirs("./samples", exist_ok=True)
ensure_dirs(args.runs_dir)
model_keys = ["A", "B"] if args.models == "both" else [args.models]
api_types = ["completions", "responses"] if args.apis == "both" else [args.apis]
tool_variants = ["chat", "responses"] if args.tools == "both" else [args.tools]
structured_opts = [True, False] if args.structured == "both" else [args.structured == "on"]
for mk in model_keys:
for api in api_types:
for tv in tool_variants:
for st in structured_opts:
run_one(mk, api, tv, st, args.task, args.runs_dir)
if __name__ == "__main__":
main()
사용 포인트
-
structured 모드일 때만
result.json을 생성하며 내용은 반드시{"filename","contents"}로 저장돼. -
contents에는 “선택한 파일의 2~3문장 요약”이 들어가도록 시스템 메시지에서 강제했어.
-
plain 모드에선 자유 형식 출력이지만, 입력과 모든 원시 응답은
input.json,outputs.json로 저장돼. -
각 단계의 콘솔 출력은 색상으로 구분돼서 입력/툴요청/출력이 한눈에 보여.
원하면 contents를 “파일 전문”으로 바꾸는 옵션 스위치도 추가해줄 수 있어.
Completions(structured)에서 ValueError list_files_in_folder is not strict. Only strict function tools can be auto-parsed.
좋은 피드백이야. 에러 원인은 chat.completions.parse를 사용할 때 툴이 strict 모드가 아니면 자동 파싱이 불가능해서야. 즉, structured 모드에서 함수 툴 정의에 strict=True가 반드시 들어가야 해.
수정 방향 요약 단계별
-
tools_variant와 structured 여부에 따라 툴 JSON에 strict를 주입하는 헬퍼를 만든다.
-
Chat 스타일일 땐 tool["function"]["strict"]=True, Responses 스타일일 땐 tool["strict"]=True를 넣는다.
-
run_chat / run_responses에서 기존 정적 TOOLS 대신 이 헬퍼로 반환된 툴을 사용한다.
아래 변경만 적용하면 돼. 나머지 로직은 그대로 둬도 동작한다.
추가 함수 및 변경점
import copy
def make_tools(tools_variant: str, structured: bool):
"""
tools_variant: 'chat' | 'responses'
structured가 True면 strict 플래그를 주입한다.
"""
base = CHAT_STYLE_TOOLS if tools_variant == "chat" else RESPONSES_STYLE_TOOLS
tools = copy.deepcopy(base)
if structured:
# Chat 스타일: "function" 블록 안에 strict
if tools_variant == "chat":
for t in tools:
if t.get("type") == "function" and "function" in t:
t["function"]["strict"] = True
# Responses 스타일: 최상위에 strict
else:
for t in tools:
if t.get("type") == "function":
t["strict"] = True
return tools
run_chat 내부에서 툴 선택 부분 교체
# 기존
# tools = CHAT_STYLE_TOOLS if tools_variant == "chat" else RESPONSES_STYLE_TOOLS
# 변경
tools = make_tools(tools_variant, structured)
run_responses 내부에서도 동일하게 교체
# 기존
# tools = RESPONSES_STYLE_TOOLS if tools_variant == "responses" else CHAT_STYLE_TOOLS
# 변경
tools = make_tools(tools_variant, structured)
추가 참고
-
이 변경으로 Completions(structured) 뿐만 아니라 Responses(structured)에서도 일관되게 엄격 파싱이 보장돼.
-
plain 모드에선 strict가 없어도 되지만, 위 헬퍼는 structured일 때만 strict를 주입하므로 기존 동작을 해치지 않아.
-
tools JSON을 동적으로 주입하므로, 새 함수만 추가하고 두 곳 호출부만 바꾸면 끝나.
이렇게 수정하면 “ValueError list_files_in_folder is not strict. Only strict function tools can be auto-parsed.” 에러는 사라지고 16개 조합 모두 정상적으로 순회할 수 있어.
그래도 계속 에러가 나네. 일단 넘어가고 json scheme을 지정하는 방식은 동작을 잘 안하네 일반 json 모드를 켜보고 싶어. 총 3가지 sweep(no structured, json_mode, structured)로 실험해보자. 예시 responses
we_did_not_specify_stop_tokens = True
try:
response = client.responses.create(
model="gpt-3.5-turbo-0125",
input=[
{"role": "system", "content": "You are a helpful assistant designed to output JSON."},
{"role": "user", "content": "Who won the world series in 2020? Please respond in the format {winner: ...}"}
],
text={"format": {"type": "json_object"}}
)
# Check if the conversation was too long for the context window, resulting in incomplete JSON
if response.status == "incomplete" and response.incomplete_details.reason == "max_output_tokens":
# your code should handle this error case
pass
# Check if the OpenAI safety system refused the request and generated a refusal instead
if response.output[0].content[0].type == "refusal":
# your code should handle this error case
# In this case, the .content field will contain the explanation (if any) that the model generated for why it is refusing
print(response.output[0].content[0]["refusal"])
# Check if the model's output included restricted content, so the generation of JSON was halted and may be partial
if response.status == "incomplete" and response.incomplete_details.reason == "content_filter":
# your code should handle this error case
pass
if response.status == "completed":
# In this case the model has either successfully finished generating the JSON object according to your schema, or the model generated one of the tokens you provided as a "stop token"
if we_did_not_specify_stop_tokens:
# If you didn't specify any stop tokens, then the generation is complete and the content key will contain the serialized JSON object
# This will parse successfully and should now contain "{"winner": "Los Angeles Dodgers"}"
print(response.output_text)
else:
# Check if the response.output_text ends with one of your stop tokens and handle appropriately
pass
except Exception as e:
# Your code should handle errors here, for example a network error calling the API
print(e)
completions
response = client.chat.completions.create(
model="gpt-4-1106-preview",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "LangChain은 무엇을 하는 라이브러리지? JSON으로 답해줘. JSON 키워드는 Description, Features, Usage로 해줘."}
],
response_format = {'type':"json_object"}
)
좋아. strict 기반의 JSON Schema 파싱은 접고, 세 모드로 스윕하도록 전체 스크립트를 업데이트했어.
스윕 축: 2모델 × 2 API(completions/responses) × 2 Tools JSON(chat-style/responses-style) × 3 모드(plain, json_mode, structured) = 총 24조합.
핵심
-
plain: 자유 출력. 전체 입력/출력 원본 JSON을 저장.
-
json_mode: 모델에게 “무조건 JSON”을 요구. 출력은 {"filename","contents"} 형식. 저장 시 result.json 생성.
-
Chat Completions: response_format={'type': 'json_object'}
-
Responses: text={"format":{"type":"json_object"}}
-
-
structured: 이전처럼 parse 사용. 여전히 모델/SDK에 따라 strict 요구로 실패할 수도 있으니 try/except로 격리. 성공 시 result.json 생성.
-
Responses API는 툴 호출이 나오면 여러 번 호출되며 매 스텝 콘솔 출력. 콘솔은 색상으로 입력(CYAN), 툴 요청(YELLOW), 출력(GREEN) 구분.
-
모든 조합은 runs/타임스탬프/모델_API_tools_mode/ 폴더에 input.json, outputs.json 저장. json_mode/structured는 result.json 추가 저장.
-
에러는 조합별로만 출력하고 나머지 계속 진행.
코드: test_24combos.py
# test_24combos.py
# Python 3.10+
# pip install --upgrade openai pydantic
import os
import json
import time
import copy
import argparse
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
from pydantic import BaseModel
from openai import OpenAI
# ===== ANSI colors =====
RESET = "\033[0m"
CYAN = "\033[36m" # input
YELLOW = "\033[33m" # tool
GREEN = "\033[32m" # output
RED = "\033[31m" # error
# ===== Model configs =====
@dataclass
class ModelConfig:
name: str
base_url: str
api_key: str
MODELS: Dict[str, ModelConfig] = {
"A": ModelConfig(
name=os.getenv("MODEL_A_NAME", "gpt-4.1-mini"),
base_url=os.getenv("OPENAI_BASE_URL_A", "https://api.openai.com/v1"),
api_key=os.getenv("OPENAI_API_KEY_A", "YOUR_KEY_FOR_MODEL_A"),
),
"B": ModelConfig(
name=os.getenv("MODEL_B_NAME", "gpt-4o-mini"),
base_url=os.getenv("OPENAI_BASE_URL_B", "https://api.openai.com/v1"),
api_key=os.getenv("OPENAI_API_KEY_B", "YOUR_KEY_FOR_MODEL_B"),
),
}
# ===== Local tools =====
def list_files_in_folder(folder_path: str) -> List[str]:
try:
return sorted([
f for f in os.listdir(folder_path)
if os.path.isfile(os.path.join(folder_path, f))
])
except Exception:
return []
def read_text_file(file_path: str) -> str:
if not os.path.exists(file_path):
return ""
try:
with open(file_path, "r", encoding="utf-8") as f:
return f.read()
except Exception:
return ""
def dispatch_tool(name: str, args: Dict[str, Any]) -> Dict[str, Any]:
if name == "list_files_in_folder":
return {"result": list_files_in_folder(args.get("folder_path", ""))}
if name == "read_text_file":
return {"result": read_text_file(args.get("file_path", ""))}
return {"error": f"Unknown tool: {name}"}
# ===== Tools JSON (two shapes) =====
CHAT_STYLE_TOOLS = [
{
"type": "function",
"function": {
"name": "list_files_in_folder",
"description": "List file names (not directories) inside a given folder path.",
"parameters": {
"type": "object",
"properties": {
"folder_path": {"type": "string", "description": "Absolute or relative folder path."}
},
"required": ["folder_path"],
"additionalProperties": False,
},
},
},
{
"type": "function",
"function": {
"name": "read_text_file",
"description": "Read a UTF-8 text file and return its content.",
"parameters": {
"type": "object",
"properties": {
"file_path": {"type": "string", "description": "Absolute or relative file path."}
},
"required": ["file_path"],
"additionalProperties": False,
},
},
},
]
RESPONSES_STYLE_TOOLS = [
{
"type": "function",
"name": "list_files_in_folder",
"description": "List file names (not directories) inside a given folder path.",
"parameters": {
"type": "object",
"properties": {
"folder_path": {"type": "string", "description": "Absolute or relative folder path."}
},
"required": ["folder_path"],
"additionalProperties": False,
},
},
{
"type": "function",
"name": "read_text_file",
"description": "Read a UTF-8 text file and return its content.",
"parameters": {
"type": "object",
"properties": {
"file_path": {"type": "string", "description": "Absolute or relative file path."}
},
"required": ["file_path"],
"additionalProperties": False,
},
},
]
# ===== Structured JSON schema (optional) =====
class TaskAnswer(BaseModel):
filename: str
contents: str
# ===== Helpers =====
def ensure_dirs(path: str): os.makedirs(path, exist_ok=True)
def write_json(path: str, obj: Any):
with open(path, "w", encoding="utf-8") as f:
json.dump(obj, f, ensure_ascii=False, indent=2)
def now_id() -> str: return time.strftime("%Y%m%d-%H%M%S")
def print_section(title: str, color: str, payload: Any):
print(color + f"\n[{title}]" + RESET)
try:
print(json.dumps(payload, ensure_ascii=False, indent=2))
except Exception:
print(payload)
def combo_label(model_key: str, api_type: str, tools_variant: str, mode: str) -> str:
return f"{model_key}_{api_type}_{tools_variant}_{mode}"
def make_tools(tools_variant: str, structured: bool) -> List[Dict[str, Any]]:
# structured에서 strict 문제를 피하기 위해 여기선 strict 주입을 끔
# (structured는 parse() 사용 시 여전히 실패할 수 있으므로 try/except로 감싼다)
base = CHAT_STYLE_TOOLS if tools_variant == "chat" else RESPONSES_STYLE_TOOLS
return copy.deepcopy(base)
# ===== Common parsing for Responses =====
def _collect_function_calls_from_output(output: List[Any]) -> List[Dict[str, Any]]:
calls: List[Dict[str, Any]] = []
for item in output or []:
itype = getattr(item, "type", None)
if itype == "function_call":
args = getattr(item, "arguments", {})
if isinstance(args, str):
try: args = json.loads(args)
except Exception: args = {}
calls.append({"call_id": getattr(item, "call_id", None), "name": getattr(item, "name", None), "arguments": args or {}})
if itype == "message":
inner = getattr(item, "tool_calls", None)
if inner:
for c in inner:
nm = getattr(getattr(c, "function", {}), "name", None) if hasattr(c, "function") else getattr(c, "name", None)
args = getattr(getattr(c, "function", {}), "arguments", {})
if isinstance(args, str):
try: args = json.loads(args)
except Exception: args = {}
calls.append({"call_id": getattr(c, "id", None), "name": nm, "arguments": args or {}})
return calls
def _collect_text_from_output(output: List[Any]) -> str:
texts: List[str] = []
for item in output or []:
if getattr(item, "type", "") == "message":
for c in getattr(item, "content", []) or []:
if getattr(c, "type", "") == "output_text":
texts.append(getattr(c, "text", ""))
if getattr(item, "type", "") == "output_text":
texts.append(getattr(item, "text", ""))
return "\n".join([t for t in texts if t])
# ===== Chat Completions runner =====
def run_chat(
cfg: ModelConfig,
tools_variant: str,
mode: str, # "plain" | "json_mode" | "structured"
user_task: str,
run_dir: str,
) -> None:
client = OpenAI(base_url=cfg.base_url, api_key=cfg.api_key)
tools = make_tools(tools_variant, structured=(mode == "structured"))
system_msg = (
"You are a helpful assistant that uses tools when needed. "
"Task: List files in ./samples, pick a text file, read it, and return a short 2-3 sentence summary. "
"When asked to answer in JSON, reply only with an object: {\"filename\": string, \"contents\": string}."
)
messages = [
{"role": "system", "content": system_msg},
{"role": "user", "content": user_task},
]
input_snapshot = {"messages": messages.copy(), "tools": tools, "mode": mode}
outputs_dump: List[Dict[str, Any]] = []
print_section("입력(Chat Completions)", CYAN, input_snapshot)
if mode == "structured":
try:
completion = client.chat.completions.parse(
model=cfg.name,
messages=messages,
tools=tools,
response_format=TaskAnswer,
tool_choice="auto",
)
outputs_dump.append(json.loads(completion.model_dump_json()))
msg = completion.choices[0].message
guard = 0
while getattr(msg, "tool_calls", None) and guard < 12:
guard += 1
tool_msgs = []
for call in msg.tool_calls:
fn_name = call.function.name
try:
fn_args = json.loads(call.function.arguments or "{}")
except Exception:
fn_args = {}
print_section("툴 요청(Chat)", YELLOW, {"name": fn_name, "arguments": fn_args})
tool_output = dispatch_tool(fn_name, fn_args)
tool_msgs.append({"role": "tool", "tool_call_id": call.id, "name": fn_name,
"content": json.dumps(tool_output, ensure_ascii=False)})
messages.append({"role": "assistant", "content": msg.content or "",
"tool_calls": [tc.model_dump() for tc in msg.tool_calls]})
messages.extend(tool_msgs)
completion = client.chat.completions.parse(
model=cfg.name, messages=messages, response_format=TaskAnswer
)
outputs_dump.append(json.loads(completion.model_dump_json()))
msg = completion.choices[0].message
parsed = getattr(msg, "parsed", None)
out_obj = parsed.model_dump() if parsed else {"filename": "", "contents": msg.content or ""}
print_section("출력(Chat structured)", GREEN, out_obj)
write_json(os.path.join(run_dir, "result.json"), out_obj)
except Exception as e:
print(RED + "\n--- ERROR(Chat structured) ---" + RESET)
print(repr(e))
elif mode == "json_mode":
# Force JSON output
resp = client.chat.completions.create(
model=cfg.name,
messages=messages,
tools=tools,
tool_choice="auto",
response_format={"type": "json_object"},
)
outputs_dump.append(json.loads(resp.model_dump_json()))
msg = resp.choices[0].message
guard = 0
while getattr(msg, "tool_calls", None) and guard < 12:
guard += 1
tool_msgs = []
for call in msg.tool_calls:
fn_name = call.function.name
try:
fn_args = json.loads(call.function.arguments or "{}")
except Exception:
fn_args = {}
print_section("툴 요청(Chat)", YELLOW, {"name": fn_name, "arguments": fn_args})
tool_output = dispatch_tool(fn_name, fn_args)
tool_msgs.append({"role": "tool", "tool_call_id": call.id, "name": fn_name,
"content": json.dumps(tool_output, ensure_ascii=False)})
messages.append({"role": "assistant", "content": msg.content or "",
"tool_calls": [tc.model_dump() for tc in msg.tool_calls]})
messages.extend(tool_msgs)
resp = client.chat.completions.create(
model=cfg.name,
messages=messages,
response_format={"type": "json_object"},
)
outputs_dump.append(json.loads(resp.model_dump_json()))
msg = resp.choices[0].message
# Try to parse JSON
result_obj = {}
try:
result_obj = json.loads(msg.content or "{}")
except Exception:
result_obj = {"filename": "", "contents": msg.content or ""}
print_section("출력(Chat json_mode)", GREEN, result_obj)
write_json(os.path.join(run_dir, "result.json"), result_obj)
else: # plain
resp = client.chat.completions.create(
model=cfg.name, messages=messages, tools=tools, tool_choice="auto"
)
outputs_dump.append(json.loads(resp.model_dump_json()))
msg = resp.choices[0].message
guard = 0
while getattr(msg, "tool_calls", None) and guard < 12:
guard += 1
tool_msgs = []
for call in msg.tool_calls:
fn_name = call.function.name
try:
fn_args = json.loads(call.function.arguments or "{}")
except Exception:
fn_args = {}
print_section("툴 요청(Chat)", YELLOW, {"name": fn_name, "arguments": fn_args})
tool_output = dispatch_tool(fn_name, fn_args)
tool_msgs.append({"role": "tool", "tool_call_id": call.id, "name": fn_name,
"content": json.dumps(tool_output, ensure_ascii=False)})
messages.append({"role": "assistant", "content": msg.content or "",
"tool_calls": [tc.model_dump() for tc in msg.tool_calls]})
messages.extend(tool_msgs)
resp = client.chat.completions.create(model=cfg.name, messages=messages)
outputs_dump.append(json.loads(resp.model_dump_json()))
msg = resp.choices[0].message
print_section("출력(Chat plain)", GREEN, {"text": msg.content})
# Save IO logs
write_json(os.path.join(run_dir, "input.json"), input_snapshot)
write_json(os.path.join(run_dir, "outputs.json"), outputs_dump)
# ===== Responses runner =====
def run_responses(
cfg: ModelConfig,
tools_variant: str,
mode: str, # "plain" | "json_mode" | "structured"
user_task: str,
run_dir: str,
) -> None:
client = OpenAI(base_url=cfg.base_url, api_key=cfg.api_key)
tools = make_tools(tools_variant, structured=(mode == "structured"))
system_inst = (
"You are a helpful assistant that uses tools when needed. "
"Task: List files in ./samples, pick a text file, read it, and return a short 2-3 sentence summary. "
"If JSON is requested, reply only with an object: {\"filename\": string, \"contents\": string}."
)
input_list: List[Dict[str, Any]] = [
{"role": "system", "content": system_inst},
{"role": "user", "content": user_task},
]
input_snapshot = {"input": input_list.copy(), "tools": tools, "mode": mode}
outputs_dump: List[Dict[str, Any]] = []
print_section("입력(Responses)", CYAN, input_snapshot)
if mode == "structured":
try:
r = client.responses.parse(
model=cfg.name,
input=input_list,
tools=tools,
text_format=TaskAnswer,
)
outputs_dump.append(json.loads(r.model_dump_json()))
print_section("출력(Responses step 1)", GREEN, json.loads(r.model_dump_json()))
input_list += list(getattr(r, "output", []) or [])
guard = 0
while guard < 12:
guard += 1
calls = _collect_function_calls_from_output(getattr(r, "output", []))
if not calls:
break
for call in calls:
print_section("툴 요청(Responses)", YELLOW, {"name": call["name"], "arguments": call["arguments"]})
result = dispatch_tool(call["name"], call["arguments"])
input_list.append({
"type": "function_call_output",
"call_id": call["call_id"],
"output": json.dumps(result, ensure_ascii=False),
})
r = client.responses.parse(
model=cfg.name, input=input_list, tools=tools, text_format=TaskAnswer
)
outputs_dump.append(json.loads(r.model_dump_json()))
print_section("출력(Responses next)", GREEN, json.loads(r.model_dump_json()))
input_list += list(getattr(r, "output", []) or [])
parsed = getattr(r, "output_parsed", None)
out_obj = parsed.model_dump() if parsed else {"filename": "", "contents": _collect_text_from_output(getattr(r, "output", []))}
print_section("출력 요약(Responses structured)", GREEN, out_obj)
write_json(os.path.join(run_dir, "result.json"), out_obj)
except Exception as e:
print(RED + "\n--- ERROR(Responses structured) ---" + RESET)
print(repr(e))
elif mode == "json_mode":
r = client.responses.create(
model=cfg.name,
input=input_list,
tools=tools,
tool_choice="auto",
text={"format": {"type": "json_object"}},
)
outputs_dump.append(json.loads(r.model_dump_json()))
print_section("출력(Responses step 1)", GREEN, json.loads(r.model_dump_json()))
input_list += list(getattr(r, "output", []) or [])
guard = 0
while guard < 12:
guard += 1
calls = _collect_function_calls_from_output(getattr(r, "output", []))
if not calls:
break
for call in calls:
print_section("툴 요청(Responses)", YELLOW, {"name": call["name"], "arguments": call["arguments"]})
result = dispatch_tool(call["name"], call["arguments"])
input_list.append({
"type": "function_call_output",
"call_id": call["call_id"],
"output": json.dumps(result, ensure_ascii=False),
})
r = client.responses.create(
model=cfg.name,
input=input_list,
tools=tools,
text={"format": {"type": "json_object"}},
)
outputs_dump.append(json.loads(r.model_dump_json()))
print_section("출력(Responses next)", GREEN, json.loads(r.model_dump_json()))
input_list += list(getattr(r, "output", []) or [])
# Try to parse JSON text
result_text = r.output_text or ""
try:
result_obj = json.loads(result_text)
except Exception:
result_obj = {"filename": "", "contents": result_text}
print_section("출력 요약(Responses json_mode)", GREEN, result_obj)
write_json(os.path.join(run_dir, "result.json"), result_obj)
else: # plain
r = client.responses.create(
model=cfg.name,
input=input_list,
tools=tools,
tool_choice="auto",
)
outputs_dump.append(json.loads(r.model_dump_json()))
print_section("출력(Responses step 1)", GREEN, json.loads(r.model_dump_json()))
input_list += list(getattr(r, "output", []) or [])
guard = 0
while guard < 12:
guard += 1
calls = _collect_function_calls_from_output(getattr(r, "output", []))
if not calls:
break
for call in calls:
print_section("툴 요청(Responses)", YELLOW, {"name": call["name"], "arguments": call["arguments"]})
result = dispatch_tool(call["name"], call["arguments"])
input_list.append({
"type": "function_call_output",
"call_id": call["call_id"],
"output": json.dumps(result, ensure_ascii=False),
})
r = client.responses.create(model=cfg.name, input=input_list, tools=tools)
outputs_dump.append(json.loads(r.model_dump_json()))
print_section("출력(Responses next)", GREEN, json.loads(r.model_dump_json()))
input_list += list(getattr(r, "output", []) or [])
print_section("출력 요약(Responses plain)", GREEN, {"text": _collect_text_from_output(getattr(r, "output", []))})
# Save IO logs
write_json(os.path.join(run_dir, "input.json"), input_snapshot)
write_json(os.path.join(run_dir, "outputs.json"), outputs_dump)
# ===== 24-combo driver =====
DEFAULT_TASK = (
"테스트: ./samples 폴더의 파일 목록을 보여주고, 텍스트 파일이 있으면 하나를 골라 내용을 읽고 2~3문장으로 요약해줘. "
"가능하다면 JSON으로 {\"filename\": string, \"contents\": string} 형태로 답해줘."
)
def run_one(model_key: str, api_type: str, tools_variant: str, mode: str, user_task: str, base_runs_dir: str):
cfg = MODELS[model_key]
label = combo_label(model_key, api_type, tools_variant, mode)
run_dir = os.path.join(base_runs_dir, label)
ensure_dirs(run_dir)
header = f"[{api_type.upper()}] model={model_key}:{cfg.name} tools={tools_variant} mode={mode}"
print("\n" + "=" * (len(header) + 2))
print(f" {header} ")
print("=" * (len(header) + 2))
try:
if api_type == "completions":
run_chat(cfg, tools_variant, mode, user_task, run_dir)
elif api_type == "responses":
run_responses(cfg, tools_variant, mode, user_task, run_dir)
else:
raise ValueError("api_type must be 'completions' or 'responses'")
except Exception as e:
print(RED + "\n--- ERROR ---" + RESET)
print(repr(e))
def main():
parser = argparse.ArgumentParser(description="Run 24 combos: 2 models × 2 APIs × 2 tool JSONs × 3 modes.")
parser.add_argument("--task", default=DEFAULT_TASK)
parser.add_argument("--models", choices=["A", "B", "both"], default="both")
parser.add_argument("--apis", choices=["completions", "responses", "both"], default="both")
parser.add_argument("--tools", choices=["chat", "responses", "both"], default="both")
parser.add_argument("--modes", choices=["plain", "json_mode", "structured", "all"], default="all")
parser.add_argument("--runs_dir", default=f"./runs/{now_id()}")
args = parser.parse_args()
os.makedirs("./samples", exist_ok=True)
ensure_dirs(args.runs_dir)
model_keys = ["A", "B"] if args.models == "both" else [args.models]
api_types = ["completions", "responses"] if args.apis == "both" else [args.apis]
tool_variants = ["chat", "responses"] if args.tools == "both" else [args.tools]
mode_list = ["plain", "json_mode", "structured"] if args.modes == "all" else [args.modes]
for mk in model_keys:
for api in api_types:
for tv in tool_variants:
for md in mode_list:
run_one(mk, api, tv, md, args.task, args.runs_dir)
if __name__ == "__main__":
main()
사용 팁
-
json_mode가 가장 안정적으로 {"filename","contents"}를 뱉어 줄 거야. structured는 SDK/모델의 strict 정책에 따라 여전히 실패할 수 있지만, 스크립트는 개별 조합별로 에러를 삼키고 계속 진행한다.
-
모든 조합의 입력/출력(raw)을 저장해서 리그레션 비교가 가능해. json_mode/structured는 result.json을 따로 저장하므로 바로 사용할 수 있어.
-
필요하면 system 메시지의 지시를 수정해서 contents에 “요약” 대신 “원문 전체”를 넣도록 바꿀 수도 있어.