[SGLang] Function Calling & Tool Use: 20+ 모델별 포맷 파서 구현

2026년 4월 9일수정: 2026년 4월 9일

들어가며

LLM이 외부 함수를 호출하는 Function Calling(Tool Use)은 이제 Agent 시스템의 핵심 인터페이스다. 문제는 모델마다 Tool Call을 표현하는 포맷이 완전히 다르다는 것이다. Qwen은 XML 태그(<tool_call>)를 쓰고, DeepSeek은 유니코드 특수 토큰(<｜tool▁calls▁begin｜>)을 쓰고, Mistral은 [TOOL_CALLS] 마커를 쓰고, Llama 4는 Python 함수 호출 문법([func(arg=val)])을 쓴다.

SGLang은 이 파편화된 포맷들을 하나의 통합 파이프라인으로 처리한다. python/sglang/srt/function_call/ 디렉토리에 24개의 모듈이 있으며, 20개 이상의 모델 포맷을 지원한다. 이 글에서는 파이프라인의 전체 구조와 핵심 구현을 코드 레벨에서 분석한다.

Tool Calling 파이프라인 구조도

LLM의 출력이 Tool Call로 변환되어 실행되기까지의 전체 흐름이다.

┌──────────────────────────────────────────────────────────────┐
│                    OpenAI-compatible Request                 │
│         (tools=[ ], tool_choice="auto" / "required")         │
└──────────────────────┬───────────────────────────────────────┘
                       │
                       ▼
┌──────────────────────────────────────────────────────────────┐
│               FunctionCallParser  초기화                      │
│  ToolCallParserEnum[tool_call_parser] → Detector 인스턴스     │
│  (예: "qwen25" → Qwen25Detector)                             │
└──────────────────────┬───────────────────────────────────────┘
                       │
                       ▼
┌──────────────────────────────────────────────────────────────┐
│                  LLM 토큰 생성 (Streaming)                    │
│  모델 출력: "<tool_call>\n{\"name\":\"get_weather\", ...}"    │
└──────────────────────┬───────────────────────────────────────┘
                       │
          ┌────────────┴────────────┐
          ▼                         ▼
  ┌───────────────┐        ┌───────────────────┐
  │  Non-Stream   │        │     Streaming     │
  │ detect_and_   │        │ parse_streaming_  │
  │ parse()       │        │ increment()       │
  └───────┬───────┘        └────────┬──────────┘
          │                         │
          └────────────┬────────────┘
                       ▼
┌──────────────────────────────────────────────────────────────┐
│          StreamingParseResult                                │
│  ├── normal_text: str  (일반 텍스트)                          │
│  └── calls: List[ToolCallItem]                               │
│       ├── tool_index: int                                    │
│       ├── name: str          ("get_weather")                 │
│       └── parameters: str    ("{\"location\":\"Tokyo\"}")    │
└──────────────────────┬───────────────────────────────────────┘
                       │
                       ▼
┌──────────────────────────────────────────────────────────────┐
│            ToolServer (MCP / Demo)                            │
│  MCPToolServer: SSE로 외부 MCP 서버 연결                      │
│  DemoToolServer: Browser, Python 내장 도구                    │
└──────────────────────────────────────────────────────────────┘

핵심 코드 분석

FunctionCallParser: 통합 진입점

FunctionCallParser는 모든 모델의 Tool Call 파싱을 하나의 인터페이스로 통합한다. 핵심은 ToolCallParserEnum 딕셔너리로, 문자열 키를 Detector 클래스에 매핑한다.

# python/sglang/srt/function_call/function_call_parser.py

class FunctionCallParser:
    ToolCallParserEnum: Dict[str, Type[BaseFormatDetector]] = {
        "deepseekv3": DeepSeekV3Detector,
        "deepseekv31": DeepSeekV31Detector,
        "deepseekv32": DeepSeekV32Detector,
        "qwen": Qwen25Detector,
        "qwen25": Qwen25Detector,
        "qwen3_coder": Qwen3CoderDetector,
        "llama3": Llama32Detector,
        "mistral": MistralDetector,
        "pythonic": PythonicDetector,
        "hermes": HermesDetector,
        "gemma4": Gemma4Detector,
        # ... 총 24개 엔트리
    }

    def __init__(self, tools: List[Tool], tool_call_parser: str):
        detector_class = self.ToolCallParserEnum.get(tool_call_parser)
        if detector_class:
            detector = detector_class()
        else:
            raise ValueError(f"Unsupported tool_call_parser: {tool_call_parser}")
        self.detector = detector

서버 시작 시 --tool-call-parser qwen25처럼 모델 포맷을 지정하면, 해당 Detector가 인스턴스화된다. 이후 모든 파싱은 이 Detector에 위임된다.

Non-streaming과 streaming 파싱은 각각 다른 메서드로 분리된다:

# python/sglang/srt/function_call/function_call_parser.py

def parse_non_stream(self, full_text: str) -> Tuple[str, list[ToolCallItem]]:
    if not self.tools:
        return full_text, []
    parsed_result = self.detector.detect_and_parse(full_text, self.tools)
    tool_call_list = parsed_result.calls
    if tool_call_list:
        return parsed_result.normal_text, tool_call_list
    else:
        return full_text, []

def parse_stream_chunk(self, chunk_text: str) -> Tuple[str, list[ToolCallItem]]:
    if not self.tools:
        return chunk_text, []
    sp_result = self.detector.parse_streaming_increment(chunk_text, self.tools)
    # ...
    return final_normal_text, final_calls

BaseFormatDetector: 스트리밍 파싱의 상태 관리

모든 Detector의 부모 클래스인 BaseFormatDetector는 스트리밍 파싱을 위한 상태 머신을 관리한다. 핵심 상태 변수들이다:

# python/sglang/srt/function_call/base_format_detector.py

class BaseFormatDetector(ABC):
    def __init__(self):
        self._buffer = ""                    # 불완전한 패턴 누적 버퍼
        self.prev_tool_call_arr: List[Dict] = []  # 완료된 tool call 정보
        self.current_tool_id: int = -1       # 현재 스트리밍 중인 tool 인덱스
        self.current_tool_name_sent: bool = False  # tool name 전송 여부
        self.streamed_args_for_tool: List[str] = []  # tool별 스트리밍된 인자

        # 서브클래스에서 오버라이드
        self.bot_token = ""    # tool call 시작 토큰
        self.eot_token = ""    # tool call 종료 토큰

스트리밍 파싱의 동작 원리를 요약하면 다음과 같다:

버퍼링: 새 텍스트가 도착하면 _buffer에 누적한다
Tool Call 감지: bot_token이 버퍼에 존재하는지 확인한다
점진적 JSON 파싱: partial_json_loads로 불완전한 JSON도 파싱한다
이름 전송 → 인자 스트리밍: tool name을 먼저 보내고, arguments를 점진적으로 diff 계산하여 전송한다

_ends_with_partial_token 메서드는 버퍼 끝에 bot_token의 일부가 걸려 있는 경우를 처리한다. 예를 들어 <tool_cal까지만 도착했을 때 이것이 <tool_call>의 시작인지 판단한다:

# python/sglang/srt/function_call/base_format_detector.py

def _ends_with_partial_token(self, buffer: str, bot_token: str) -> int:
    for i in range(1, min(len(buffer) + 1, len(bot_token))):
        if bot_token.startswith(buffer[-i:]):
            return i
    return 0

모델별 Detector 구현 패턴

각 Detector는 BaseFormatDetector를 상속하고, 최소 3개의 메서드를 구현한다: has_tool_call, detect_and_parse, structure_info. 모델별로 포맷 차이가 크기 때문에 구현 전략이 다르다.

Qwen 2.5/3 -- XML 태그 기반으로, 기본 스트리밍 로직을 재사용하면서 종료 태그 처리만 커스텀한다:

# python/sglang/srt/function_call/qwen25_detector.py

class Qwen25Detector(BaseFormatDetector):
    def __init__(self):
        super().__init__()
        self.bot_token = "<tool_call>\n"
        self.eot_token = "\n</tool_call>"

    def detect_and_parse(self, text, tools):
        pattern = rf"{re.escape(self.bot_token)}(.*?){re.escape(self.eot_token)}"
        match_result_list = re.findall(pattern, text, re.DOTALL)
        calls = []
        for match_result in match_result_list:
            parsed_call = json.loads(match_result.strip())
            calls.extend(self.parse_base_json(parsed_call, tools))
        return StreamingParseResult(normal_text=normal_text, calls=calls)

DeepSeek V3 -- 유니코드 특수 토큰과 JSON 코드 블록을 조합한 독자적 포맷이다. 정규식으로 함수명과 인자를 분리한다:

# python/sglang/srt/function_call/deepseekv3_detector.py

class DeepSeekV3Detector(BaseFormatDetector):
    def __init__(self):
        super().__init__()
        self.bot_token = "<｜tool▁calls▁begin｜>"
        self.eot_token = "<｜tool▁calls▁end｜>"
        self.func_detail_regex = (
            r"<｜tool▁call▁begin｜>(.*)<｜tool▁sep｜>(.*)\n```json\n(.*)\n```"
            r"<｜tool▁call▁end｜>"
        )

Llama 4 (Pythonic) -- JSON이 아닌 Python 함수 호출 문법을 파싱해야 하므로, 정규식과 ast.literal_eval을 조합한다. structural_tag를 지원하지 않는 유일한 포맷이기도 하다:

# python/sglang/srt/function_call/pythonic_detector.py

class PythonicDetector(BaseFormatDetector):
    # 포맷: [tool1(arg1=val1, arg2=val2), tool2(arg1=val3)]
    def __init__(self):
        super().__init__()
        self.tool_call_regex = re.compile(
            r"\[([a-zA-Z]+\w*\(([a-zA-Z]+\w*=.*,\s*)*"
            r"([a-zA-Z]+\w*=.*\s)?\),\s*)*"
            r"([a-zA-Z]+\w*\(([a-zA-Z]+\w*=.*,\s*)*"
            r"([a-zA-Z]+\w*=.*\s*)?\)\s*)+\]",
            re.DOTALL,
        )

Constrained Generation: structural_tag

FunctionCallParser.get_structure_constraint는 tool_choice 설정에 따라 LLM의 출력을 제한하는 제약 조건을 생성한다. tool_choice="auto"이면 structural_tag, "required"이면 json_schema 제약을 적용한다.

# python/sglang/srt/function_call/function_call_parser.py

def get_structure_constraint(self, tool_choice, parallel_tool_calls=True):
    if (self.detector.supports_structural_tag()
        and tool_choice == "auto"
        and (any(tool.function.strict for tool in self.tools)
             or self.tool_strict_level >= ToolStrictLevel.FUNCTION)):
        tag = self.get_structure_tag()
        return ("structural_tag", tag)
    elif tool_choice == "required" or isinstance(tool_choice, ToolChoice):
        json_schema = get_json_schema_constraint(
            self.tools, tool_choice, parallel_tool_calls=parallel_tool_calls)
        return ("json_schema", json_schema)

structural_tag는 각 Detector의 structure_info()가 반환하는 begin/end/trigger 패턴으로 구성된다. 예를 들어 Qwen25의 경우 begin이 <tool_call>\n{"name":"get_weather", "arguments":, end가 }\n</tool_call>이 되어, LLM이 이 구조를 벗어나는 출력을 생성하지 못하게 한다.

Tool Server: MCP 기반 함수 실행

파싱된 Tool Call은 ToolServer를 통해 실제로 실행된다. SGLang은 MCP(Model Context Protocol) 표준을 지원한다.

# python/sglang/srt/entrypoints/openai/tool_server.py

class MCPToolServer(ToolServer):
    async def add_tool_server(self, server_url: str):
        tool_urls = server_url.split(",")
        for url in tool_urls:
            url = f"http://{url}/sse"
            initialize_response, list_tools_response = (
                await list_server_and_tools(url))
            # MCP 서버의 도구 목록을 Harmony 포맷으로 변환
            tool_from_mcp = ToolNamespaceConfig(
                name=initialize_response.serverInfo.name,
                tools=[ToolDescription.new(
                    name=tool.name,
                    description=tool.description,
                    parameters=tool.inputSchema,
                ) for tool in list_tools_response.tools],
            )

MCPToolServer는 SSE(Server-Sent Events)로 외부 MCP 서버에 연결하고, DemoToolServer는 내장 Browser/Python 도구를 제공한다.

모델별 Tool Calling 포맷 비교

Parser 키	Detector 클래스	시작 토큰	인자 포맷	Parallel 지원
`qwen` / `qwen25`	Qwen25Detector	`<tool_call>\n`	JSON `{"name":..., "arguments":...}`	O
`qwen3_coder`	Qwen3CoderDetector	`<tool_call>\n`	JSON	O
`deepseekv3`	DeepSeekV3Detector	`<｜tool▁calls▁begin｜>`	JSON code block (```json)	O
`deepseekv31`	DeepSeekV31Detector	`<｜tool▁calls▁begin｜>`	JSON (no code block)	O
`deepseekv32`	DeepSeekV32Detector	`<｜tool▁calls▁begin｜>`	JSON	O
`llama3`	Llama32Detector	`<\|python_tag\|>`	JSON / Python dict	제한적
`pythonic`	PythonicDetector	`[func(`	Python 함수 호출 문법	O
`mistral`	MistralDetector	`[TOOL_CALLS]`	JSON array 또는 `[ARGS]` compact	O
`hermes`	HermesDetector	`<tool_call>`	JSON	O
`gemma4`	Gemma4Detector	`<\|tool_call\|>`	커스텀 KV 포맷 (`<\|"\|>` 구분자)	O
`glm` / `glm45`	Glm4MoeDetector	모델 종속	JSON	O
`glm47`	Glm47MoeDetector	모델 종속	JSON	O
`gpt-oss`	GptOssDetector	모델 종속	JSON	O
`kimi_k2`	KimiK2Detector	모델 종속	JSON	O
`lfm2`	Lfm2Detector	모델 종속	JSON	O
`mimo`	MiMoDetector	모델 종속	JSON	O
`minimax-m2`	MinimaxM2Detector	모델 종속	JSON	O
`step3`	Step3Detector	모델 종속	JSON	O
`step3p5`	Qwen3CoderDetector	`<tool_call>\n`	JSON	O
`trinity`	TrinityDetector	모델 종속	JSON	O
`interns1`	InternlmDetector	모델 종속	JSON	O
`gigachat3`	GigaChat3Detector	모델 종속	JSON	O

24개 엔트리, 21개 고유 Detector 클래스가 등록되어 있다. qwen/qwen25는 같은 Detector를 공유하고, step3p5는 Qwen3CoderDetector를 재사용한다.

설계 포인트 정리

1. 전략 패턴으로 포맷 분리: ToolCallParserEnum 딕셔너리가 문자열 키를 Detector 클래스에 매핑한다. 새 모델 포맷을 추가하려면 Detector 클래스 하나와 딕셔너리 엔트리 하나만 추가하면 된다.

2. 스트리밍 상태 머신: BaseFormatDetector가 _buffer, current_tool_id, current_tool_name_sent 등의 상태를 관리하면서 점진적 JSON 파싱을 수행한다. 대부분의 JSON 기반 Detector는 이 기본 구현을 그대로 상속받는다.

3. Constrained Generation 통합: 파싱만이 아니라, structural_tag와 json_schema를 통해 LLM이 올바른 포맷으로 출력하도록 생성 단계에서도 제약을 건다.

4. MCP 표준 지원: MCPToolServer가 외부 도구 서버와 SSE로 통신하여, 파싱된 Tool Call을 실제로 실행한다.

참고

SGLang 의 다른글

이전글 [SGLang] gRPC 서버: 분산 추론을 위한 고성능 통신 계층
현재글 : [SGLang] Function Calling & Tool Use: 20+ 모델별 포맷 파서 구현
다음글 [SGLang] 음성 인식 & ASR 통합: Whisper, Qwen3-ASR 어댑터 구현