vLLM 툴 호출 예제
vLLM(GPT-OSS-120B) + openai responses 파이썬 패키지를 이용해서 통신하는 예제 작성해줘. 이 때 tool calling을 해야해(harmony 형식 지원). 다음의 tool 을 지원하도록 해줘 1. 폴더 내의 파일 리스트 조회 2. 단일 파일 읽기
좋아요! vLLM으로 GPT-OSS-120B를 띄우고, Python용 OpenAI SDK의 Responses API로 “harmony 형식 + tool calling”을 사용하는 최소·완전 예제를 만들어 드릴게요. 요구하신 두 가지 툴(폴더 목록 조회, 단일 파일 읽기)을 정의하고, 모델이 툴을 호출하면 실제 함수를 실행한 뒤 결과를 다시 모델에 넘겨 최종 답을 받는 구조입니다.
아래 순서대로 진행하면 됩니다.
- vLLM 서버 실행(오픈AI 호환 엔드포인트)
- vLLM은 OpenAI-compatible 서버를 제공하므로 SDK의
base_url만 바꿔서 쓸 수 있습니다. 예시처럼 gpt-oss-120b 모델을 서빙하세요. OpenAI Cookbook
# vLLM 설치 및 서버 기동(공식 가이드에 맞춤)
# 참고: gpt-oss는 vLLM에서 Responses API와 Chat Completions API 둘 다 지원합니다.
vllm serve openai/gpt-oss-120b
# 서버는 기본적으로 http://localhost:8000/v1 에서 OpenAI 호환 API를 엽니다.
- 핵심 아이디어(툴 호출 루프)
-
Responses API로 요청을 보내면
output배열에 항목들이 들어오는데, 그중type == "function_call"인 항목이 있으면 해당 함수를 로컬에서 실행하고, 결과를tool_outputs로 다시responses.create에 넘겨 후속 추론을 이어갑니다.previous_response_id를 사용해 직전 턴과 연결합니다. OpenAI 플랫폼+1 -
GPT-OSS 계열은 Harmony 응답 포맷을 사용하도록 학습되어 있으며, vLLM + OpenAI SDK 조합을 쓰면 별도 포맷 처리 없이 그대로 동작합니다. 직접 토큰 레벨 제어가 필요할 때만 Harmony SDK를 사용하면 됩니다. OpenAI Cookbook
-
vLLM은 Responses API와 툴 호출을 지원하므로 이 코드가 그대로 작동합니다. OpenAI Cookbook
- 파이썬 예제 코드(툴 2종: 폴더 목록, 파일 읽기)
# pip install openai
# 환경: vLLM OpenAI-compatible 서버가 localhost:8000/v1 에서 실행 중이어야 합니다.
import os
import json
from pathlib import Path
from typing import List, Dict, Any
from openai import OpenAI
# 0) OpenAI 클라이언트: vLLM 호환 서버로 지정
client = OpenAI(
base_url="http://localhost:8000/v1", # vLLM OpenAI-compatible 서버
api_key="EMPTY" # vLLM 기본 토큰(검증 안함). 필요시 임의 문자열
)
# 1) 실제로 실행될 로컬 함수들(툴 구현)
def list_directory(path: str) -> Dict[str, Any]:
p = Path(path).expanduser().resolve()
if not p.exists() or not p.is_dir():
return {"ok": False, "error": f"Directory not found: {p}"}
items = []
for child in sorted(p.iterdir()):
try:
items.append({
"name": child.name,
"is_dir": child.is_dir(),
"size": None if child.is_dir() else child.stat().st_size
})
except Exception as e:
items.append({"name": child.name, "error": str(e)})
return {"ok": True, "path": str(p), "items": items}
def read_file(path: str, max_bytes: int = 200_000) -> Dict[str, Any]:
p = Path(path).expanduser().resolve()
if not p.exists() or not p.is_file():
return {"ok": False, "error": f"File not found: {p}"}
try:
# 텍스트 파일 가정. 이진 파일이면 깨질 수 있음 → errors='replace'
data = p.read_text(encoding="utf-8", errors="replace")
if len(data) > max_bytes:
data = data[:max_bytes] + "\n... [truncated]"
return {"ok": True, "path": str(p), "content": data}
except Exception as e:
return {"ok": False, "error": str(e)}
# 2) 툴 스키마 정의(Harmony/Responses용 function tool)
tools: List[Dict[str, Any]] = [
{
"type": "function",
"function": {
"name": "list_directory",
"description": "List files and folders in a directory.",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "Target directory path"}
},
"required": ["path"],
"additionalProperties": False
}
}
},
{
"type": "function",
"function": {
"name": "read_file",
"description": "Read a text file from disk.",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "Target file path"},
"max_bytes": {
"type": "integer",
"description": "Max characters to return",
"default": 200000
}
},
"required": ["path"],
"additionalProperties": False
}
}
}
]
# 로컬 함수 디스패처
FUNCTIONS = {
"list_directory": list_directory,
"read_file": read_file,
}
def extract_function_calls(response_obj: Any) -> List[Dict[str, Any]]:
"""
Responses API의 응답에서 function_call 항목을 추출.
SDK 버전마다 타입 클래스가 달라질 수 있으므로 dict로 안전하게 처리.
각 항목: { "call_id": str, "name": str, "arguments": dict }
"""
data = response_obj.model_dump() if hasattr(response_obj, "model_dump") else response_obj
calls = []
for item in data.get("output", []):
if item.get("type") == "function_call":
name = item.get("name")
call_id = item.get("call_id")
# arguments는 dict 이거나 JSON string일 수 있음
raw_args = item.get("arguments", {})
if isinstance(raw_args, str):
try:
raw_args = json.loads(raw_args)
except Exception:
raw_args = {}
calls.append({"call_id": call_id, "name": name, "arguments": raw_args})
return calls
def run_tool_call(name: str, arguments: Dict[str, Any]) -> str:
"""
해당 툴을 실제 실행하고, 문자열(JSON)로 결과를 반환.
모델로 돌려보낼 때는 문자열이어야 합니다.
"""
fn = FUNCTIONS.get(name)
if not fn:
return json.dumps({"ok": False, "error": f"Unknown function: {name}"})
try:
result = fn(**arguments)
return json.dumps(result, ensure_ascii=False)
except TypeError as e:
# 잘못된 인자 등
return json.dumps({"ok": False, "error": f"Bad arguments for {name}: {e}"}, ensure_ascii=False)
except Exception as e:
return json.dumps({"ok": False, "error": str(e)}, ensure_ascii=False)
def respond_with_tools(user_prompt: str, model: str = "openai/gpt-oss-120b") -> str:
"""
1) 사용자 입력으로 응답 생성 시도
2) 툴 호출이 나오면 실제 실행 → tool_outputs 로 재호출
3) 더 이상 툴 호출이 없을 때 최종 텍스트를 반환
"""
prev_id = None
final_text = None
while True:
if prev_id is None:
# 첫 턴: 사용자 입력을 보냄
resp = client.responses.create(
model=model,
tools=tools,
input=user_prompt
)
else:
# 두 번째 턴부터는 tool_outputs만 넘기고 이어서 추론
resp = client.responses.create(
model=model,
tools=tools,
previous_response_id=prev_id,
tool_outputs=tool_outputs # 직전 루프에서 준비
)
# 툴 호출이 있는지 확인
calls = extract_function_calls(resp)
if not calls:
# 최종 답변
# Responses SDK는 편의 프로퍼티로 output_text 를 제공합니다.
try:
final_text = resp.output_text
except Exception:
# 호환성 대비
data = resp.model_dump() if hasattr(resp, "model_dump") else resp
# output 배열에서 assistant 텍스트를 합치기
chunks = []
for item in data.get("output", []):
if item.get("type") == "message":
for c in item.get("content", []):
if c.get("type") == "output_text":
chunks.append(c.get("text", ""))
final_text = "\n".join(chunks) if chunks else json.dumps(data)
break
# 하나 이상 툴 호출 → 모두 실행 후 결과를 묶어서 넘김
tool_outputs = []
for call in calls:
result_str = run_tool_call(call["name"], call["arguments"])
tool_outputs.append({
"tool_call_id": call["call_id"],
"output": result_str
})
prev_id = resp.id # 이어지는 턴에서 previous_response_id로 연결
return final_text or ""
if __name__ == "__main__":
# 사용 예시(원하는 경로로 바꾸세요)
prompt = (
"현재 작업 폴더의 파일을 보여주고(./), "
"그중 README.md가 있으면 내용을 읽어서 요약해줘."
)
answer = respond_with_tools(prompt)
print("\n=== Assistant ===\n")
print(answer)
- 포인트 요약
-
vLLM는 OpenAI-compatible 서버로 Responses API를 그대로 지원합니다. 즉
base_url만 바꿔서client.responses.create(...)를 호출하면 됩니다. OpenAI Cookbook -
툴 스키마는
{"type": "function", "function": {name, description, parameters}}형태로 정의합니다. GPT-OSS는 Harmony 포맷을 이해하며 vLLM 경유 시 별도 처리 없이 툴 호출이 동작합니다. OpenAI Cookbook -
모델이
function_call을 생성하면, 각 항목의call_id,name,arguments를 꺼내 실제 코드를 실행하고, 결과를tool_outputs=[{"tool_call_id": ..., "output": "..."}]로 다시responses.create에 넘겨 한 턴을 이어갑니다.previous_response_id로 직전 턴을 연결합니다. OpenAI 플랫폼+1 -
gpt-oss를 vLLM로 직접 샘플링할 때는 Harmony SDK로 프롬프트/스톱토큰을 관리해야 하지만, OpenAI-compatible 서버 + SDK 경로에서는 신경 쓸 필요가 거의 없습니다. OpenAI Cookbook
필요하면 위 코드에 스트리밍(stream=True) 처리나, 바이너리 파일 감지/차단 같은 보강 로직도 덧대드릴게요.
대화를 두 턴은 했으면 좋겠어. README.md를 읽었으니 다음엔 requirements.txt를 읽자
좋아요. “두 턴 대화”가 자연스럽게 이어지도록, 첫 턴에서 README.md를 읽고 요약한 뒤, 둘째 턴에서 requirements.txt를 읽는 구조를 그대로 보여주는 예제를 드릴게요. 핵심은 Responses API에서 previous_response_id로 대화 문맥을 잇고, 툴 호출(function_call)이 올 때마다 실제 함수를 실행해 tool_outputs로 되돌려주는 루프를 매 턴마다 돌리는 것입니다.
아래 순서대로 보면 됩니다.
-
vLLM(OpenAI-compatible) 서버가 localhost:8000/v1에서 실행 중이라고 가정합니다.
-
두 개의 툴(list_directory, read_file)을 정의합니다.
-
Conversation 클래스로 “턴 단위” 대화 실행기를 만들고, 턴마다 툴 호출 루프를 처리합니다.
-
main에서 실제 두 턴을 수행합니다: 1턴(README.md) → 2턴(requirements.txt).
# pip install openai
# 사전 준비: vLLM(OpenAI-compatible) 서버를 http://localhost:8000/v1 에서 가동
# 예) vllm serve openai/gpt-oss-120b
import os
import json
from pathlib import Path
from typing import List, Dict, Any
from openai import OpenAI
# ===== 0) OpenAI 클라이언트: vLLM 호환 서버로 지정 =====
client = OpenAI(
base_url="http://localhost:8000/v1",
api_key="EMPTY" # vLLM 기본값. 필요 시 임의 문자열로 대체
)
# ===== 1) 실제로 실행될 로컬 함수들(툴 구현) =====
def list_directory(path: str) -> Dict[str, Any]:
p = Path(path).expanduser().resolve()
if not p.exists() or not p.is_dir():
return {"ok": False, "error": f"Directory not found: {p}"}
items = []
for child in sorted(p.iterdir()):
try:
items.append({
"name": child.name,
"is_dir": child.is_dir(),
"size": None if child.is_dir() else child.stat().st_size
})
except Exception as e:
items.append({"name": child.name, "error": str(e)})
return {"ok": True, "path": str(p), "items": items}
def read_file(path: str, max_bytes: int = 200_000) -> Dict[str, Any]:
p = Path(path).expanduser().resolve()
if not p.exists() or not p.is_file():
return {"ok": False, "error": f"File not found: {p}"}
try:
text = p.read_text(encoding="utf-8", errors="replace")
if len(text) > max_bytes:
text = text[:max_bytes] + "\n... [truncated]"
return {"ok": True, "path": str(p), "content": text}
except Exception as e:
return {"ok": False, "error": str(e)}
# ===== 2) 툴 스키마(Harmony/Responses용 function tool) =====
TOOLS: List[Dict[str, Any]] = [
{
"type": "function",
"function": {
"name": "list_directory",
"description": "List files and folders in a directory.",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "Target directory path"}
},
"required": ["path"],
"additionalProperties": False
}
}
},
{
"type": "function",
"function": {
"name": "read_file",
"description": "Read a text file from disk.",
"parameters": {
"type": "object",
"properties": {
"path": {"type": "string", "description": "Target file path"},
"max_bytes": {
"type": "integer",
"description": "Max characters to return",
"default": 200000
}
},
"required": ["path"],
"additionalProperties": False
}
}
}
]
FUNCTIONS = {"list_directory": list_directory, "read_file": read_file}
# ===== 3) 툴 호출 유틸 =====
def extract_function_calls(resp_obj: Any) -> List[Dict[str, Any]]:
"""
Responses API 응답에서 function_call들을 추출해
[{call_id, name, arguments}, ...] 형태로 반환
"""
data = resp_obj.model_dump() if hasattr(resp_obj, "model_dump") else resp_obj
calls = []
for item in data.get("output", []):
if item.get("type") == "function_call":
name = item.get("name")
call_id = item.get("call_id")
raw_args = item.get("arguments", {})
if isinstance(raw_args, str):
try:
raw_args = json.loads(raw_args)
except Exception:
raw_args = {}
calls.append({"call_id": call_id, "name": name, "arguments": raw_args})
return calls
def run_tool_call(name: str, arguments: Dict[str, Any]) -> str:
"""
해당 툴을 실제 실행하고 결과를 JSON 문자열로 반환
"""
fn = FUNCTIONS.get(name)
if not fn:
return json.dumps({"ok": False, "error": f"Unknown function: {name}"}, ensure_ascii=False)
try:
result = fn(**arguments)
return json.dumps(result, ensure_ascii=False)
except TypeError as e:
return json.dumps({"ok": False, "error": f"Bad arguments for {name}: {e}"}, ensure_ascii=False)
except Exception as e:
return json.dumps({"ok": False, "error": str(e)}, ensure_ascii=False)
def get_output_text(resp_obj: Any) -> str:
"""
최종 어시스턴트 텍스트를 꺼내는 헬퍼
"""
try:
return resp_obj.output_text
except Exception:
data = resp_obj.model_dump() if hasattr(resp_obj, "model_dump") else resp_obj
chunks = []
for item in data.get("output", []):
if item.get("type") == "message":
for c in item.get("content", []):
if c.get("type") == "output_text":
chunks.append(c.get("text", ""))
return "\n".join(chunks)
# ===== 4) 대화 실행기: 턴 단위로 previous_response_id를 이어감 =====
class Conversation:
def __init__(self, client: OpenAI, model: str, tools: List[Dict[str, Any]]):
self.client = client
self.model = model
self.tools = tools
self.last_response_id = None # 마지막 assistant 응답의 id
def run_turn(self, user_input: str) -> str:
"""
한 턴을 수행:
1) 사용자 입력을 모델에 보냄(문맥 연결: previous_response_id 사용)
2) function_call이 있으면 실제 실행 → tool_outputs로 재호출
3) function_call이 더 이상 없으면 최종 텍스트 반환
"""
# 1) 사용자 입력 전송
if self.last_response_id is None:
resp = self.client.responses.create(
model=self.model,
tools=self.tools,
input=user_input
)
else:
resp = self.client.responses.create(
model=self.model,
tools=self.tools,
previous_response_id=self.last_response_id,
input=user_input
)
# 2) 툴 호출 루프
while True:
calls = extract_function_calls(resp)
if not calls:
break
tool_outputs = []
for call in calls:
result = run_tool_call(call["name"], call["arguments"])
tool_outputs.append({
"tool_call_id": call["call_id"],
"output": result
})
resp = self.client.responses.create(
model=self.model,
tools=self.tools,
previous_response_id=resp.id,
tool_outputs=tool_outputs
)
# 3) 최종 텍스트와 응답 id 갱신
text = get_output_text(resp)
self.last_response_id = resp.id
return text
# ===== 5) 데모: 두 턴 대화 =====
if __name__ == "__main__":
conv = Conversation(client, model="openai/gpt-oss-120b", tools=TOOLS)
# 첫 번째 턴: README.md를 읽고 요약 요청
turn1_user = (
"프로젝트 루트(./)의 파일 목록을 보고 README.md가 있으면 읽어서 핵심을 요약해줘. "
"없으면 왜 없는지도 알려줘."
)
turn1_assistant = conv.run_turn(turn1_user)
print("\n=== [TURN 1] Assistant ===\n")
print(turn1_assistant)
# 두 번째 턴: requirements.txt 읽기
turn2_user = (
"좋아. 이제 같은 폴더에서 requirements.txt를 찾아서 내용을 읽고, "
"주요 패키지와 대략적 버전을 bullet로 정리해줘."
)
turn2_assistant = conv.run_turn(turn2_user)
print("\n=== [TURN 2] Assistant ===\n")
print(turn2_assistant)
사용 팁과 체크포인트
-
폴더 경로 기준은 현재 작업 디렉터리(./)입니다. 다른 경로를 쓰려면 프롬프트 또는 툴 인자(path)에 명시하세요.
-
모델이 스스로 list_directory를 먼저 호출해 파일 존재 여부를 확인한 다음 read_file을 호출하도록 의도를 유도하고자, 사용자 프롬프트에 “목록을 보고 … 읽어줘”라고 써두었습니다.
-
두 번째 턴에서는 previous_response_id로 문맥이 이어지므로, 같은 폴더를 바로 가정하고 requirements.txt만 요청하면 됩니다.
-
read_file은 텍스트 파일 전용이며, 큰 파일은 max_bytes로 절단합니다. 이진 파일은 깨질 수 있으니 필요하면 바이너리 감지 로직을 추가하세요.
-
스트리밍이 필요하면 responses.create(stream=True)로 바꾸고, 스트리밍 청크를 처리한 뒤 동일한 툴 호출 루프를 적용하면 됩니다.
이 스크립트를 그대로 실행하면, 실제 로컬 디렉터리의 README.md → requirements.txt 순서로 읽어서 요약/정리하는 두 턴 대화가 진행됩니다.
여기에서보면 과거 대화를 같이 안 보내주고 저번 대화 id만 보내는 것 같은데 이런게 vLLM에도 지원돼?
짧은 답: 네, 지원돼요. vLLM은 /v1/responses 엔드포인트를 갖고 previous_response_id를 받아 “직전 응답만” 넘겨도 문맥을 잇는 방식을 처리합니다. 다만 기본값에선 서버가 대화 상태를 저장하지 않도록 되어 있어, 상태 저장을 켜주거나(권장) 아니면 매 턴 전체 맥락을 직접 다시 보내야 합니다. OpenAI Cookbook+2
생각 순서
-
무엇이 가능한가: vLLM은 OpenAI Responses API와 호환되며,
previous_response_id필드를 프로토콜에서 받아 처리합니다. 즉, “직전 응답의 id만” 전송해도 체이닝이 됩니다. OpenAI Cookbook+1 -
왜 저장 옵션이 필요한가: OpenAI의 Responses API는 기본이
store=True라서 서버가 과거 턴을 내부 저장소에 보관합니다. vLLM은 기본에선 이를 꺼둔 상태이므로, 저장을 활성화하지 않으면store=True요청에 오류가 나거나, 저장 없이 동작해previous_response_id의 이점(서버측 히든 상태 재사용)을 못 누릴 수 있습니다. 환경변수VLLM_ENABLE_RESPONSES_API_STORE=1로 저장 기능을 켜면 OpenAI와 유사하게 동작합니다. 대안으로 클라이언트에서store=False로 보내고 매 턴 전체 히스토리를 직접 재전송할 수도 있습니다. VLLM Docs+2
빠른 설정 예시
서버(상태 저장 ON):
VLLM_ENABLE_RESPONSES_API_STORE=1 vllm serve openai/gpt-oss-120b --port 8000
클라이언트(두 턴 체인: 과거 대화 본문 미전송, id만 전달):
from openai import OpenAI
client = OpenAI(base_url="http://localhost:8000/v1", api_key="EMPTY")
# 턴1
r1 = client.responses.create(
model="openai/gpt-oss-120b",
input="프로젝트 루트의 README.md를 읽고 핵심만 요약해줘."
# store=True는 Responses API 기본값. vLLM에선 위 환경변수로 활성화 필요
)
print(r1.output_text)
# 턴2: 과거 본문은 안 보내고, 직전 응답의 id만 연결
r2 = client.responses.create(
model="openai/gpt-oss-120b",
previous_response_id=r1.id,
input="좋아. 이번엔 requirements.txt를 읽어서 주요 패키지와 버전을 정리해줘."
)
print(r2.output_text)
만약 서버 상태 저장을 끄고 싶다면
- 매 요청에
store=False로 보내세요. 이 경우 서버가 상태를 안 가지므로,previous_response_id를 써도 실질적 이득이 없고 스스로 히스토리를 재전송해야 합니다. VLLM Docs
주의할 점
- 멀티 인스턴스/재시작 상황에선 저장소가 꺼져 있으면 체인 컨텍스트가 유실됩니다. vLLM 커뮤니티에서도 “Responses API의 풀 기능을 쓰려면 store/message 스토어가 필요”하다는 논의가 있습니다. 운영 환경에선 저장을 켜거나 외부 영속 스토어 전략을 함께 고려하세요. GitHub
핵심 요약
-
가능: vLLM은
previous_response_id체이닝을 받습니다. VLLM Docs -
권장:
VLLM_ENABLE_RESPONSES_API_STORE=1로 상태 저장을 켜서 “과거 대화 본문 무전송 + id만 전달” 패턴을 그대로 사용하세요. VLLM Docs -
대안: 저장을 끌 땐
store=False로 보내고, 히스토리는 클라이언트가 재전송합니다. VLLM Docs
툴 다 빼고 간단히 테스트할 curl 예제만 만들어줘
원하는 것만 딱! 툴 없이 vLLM(OpenAI-compatible) Responses API를 curl로 두 턴 테스트하는 최소 예제입니다. 서버는 http://localhost:8000/v1에서 openai/gpt-oss-120b를 서빙한다고 가정합니다.
- 단발성 호출(연결 없음)
curl -s http://localhost:8000/v1/responses \
-H "Authorization: Bearer EMPTY" \
-H "Content-Type: application/json" \
-d '{
"model": "openai/gpt-oss-120b",
"input": "한 줄 농담 하나 해줘."
}' | jq -r '.output_text'
- 두 턴 대화(직전 응답 ID(previous_response_id)로 체인)
# 턴 1
RESP1=$(curl -s http://localhost:8000/v1/responses \
-H "Authorization: Bearer EMPTY" \
-H "Content-Type: application/json" \
-d '{
"model": "openai/gpt-oss-120b",
"input": "프로젝트 루트의 README.md를 읽었다고 치고, 한 줄로 요약해줘."
}')
echo "TURN1:"; echo "$RESP1" | jq -r '.output_text'
# 직전 응답의 id 추출
RID=$(echo "$RESP1" | jq -r '.id')
# 턴 2: 과거 본문은 보내지 않고 id만 연결
curl -s http://localhost:8000/v1/responses \
-H "Authorization: Bearer EMPTY" \
-H "Content-Type: application/json" \
-d "{
\"model\": \"openai/gpt-oss-120b\",
\"previous_response_id\": \"$RID\",
\"input\": \"좋아. 이제 requirements.txt를 읽었다고 치고, 주요 패키지만 bullet로 뽑아줘.\"
}" | jq -r '.output_text'
주의
-
vLLM에서
previous_response_id체이닝을 쓰려면 서버가 응답 저장을 지원해야 합니다. 서버 기동 전에 상태 저장을 켜두었다면 위 예제가 그대로 작동하고, 저장이 꺼져 있다면 매 턴 전체 맥락을 직접 보내야 합니다. -
토큰 검증을 하지 않는 기본 설정이라면 Authorization 헤더의 값은 임의 문자열(예:
EMPTY)이면 됩니다. -
모델 이름은
vllm serve시 지정한 이름과 일치해야 합니다. 예)vllm serve openai/gpt-oss-120b→ 요청 바디의"model": "openai/gpt-oss-120b".