본문으로 건너뛰기

[SGLang] Reasoning & Code Completion Parser: 추론 및 코드 파서

들어가며

최근 LLM은 Chain-of-Thought(CoT) 추론과 FIM(Fill-in-the-Middle) 코드 완성을 지원한다. SGLang의 parser/ 패키지는 이러한 출력 형식을 파싱하여 추론 내용(reasoning)과 일반 응답(content)을 분리하고, 코드 완성 프롬프트를 올바르게 구성한다.

구조도

parser/
├── reasoning_parser.py      ── 추론 출력 파싱 (think 태그)
├── code_completion_parser.py── FIM 코드 완성 프롬프트
├── conversation.py          ── 대화 템플릿 관리
├── harmony_parser.py        ── 조화 파서 (통합)
└── jinja_template_utils.py  ── Jinja 템플릿 유틸리티

[추론 파싱 흐름]
모델 출력:  <think>추론 내용</think>실제 응답
                │                        │
                ▼                        ▼
        reasoning_text           normal_text

핵심 코드 분석

BaseReasoningFormatDetector: 추론 형식 감지기

추론 파서의 핵심은 BaseReasoningFormatDetector다. think_start_tokenthink_end_token 사이의 내용을 추론으로, 나머지를 일반 텍스트로 분류한다.

class BaseReasoningFormatDetector:
    def __init__(self, think_start_token, think_end_token,
                 force_reasoning=False, stream_reasoning=True,
                 tool_start_token=None, continue_final_message=False,
                 previous_content=""):
        self.think_start_token = think_start_token
        self.think_end_token = think_end_token
        self._in_reasoning = force_reasoning
        self.stream_reasoning = stream_reasoning
        self._buffer = ""

continue_final_message=True면 이전 대화에서 이미 시작된 추론 블록을 이어서 파싱할 수 있다.

일괄 파싱: detect_and_parse

전체 텍스트를 한 번에 파싱한다. think_end_token을 기준으로 추론과 일반 텍스트를 분리한다.

def detect_and_parse(self, text: str) -> StreamingParseResult:
    in_reasoning = self._in_reasoning or self.think_start_token in text
    if not in_reasoning:
        return StreamingParseResult(normal_text=text)

    processed_text = text.replace(
        self.think_start_token + self.think_start_self_label, "").strip()

    if self.think_end_token in processed_text:
        splits = processed_text.split(self.think_end_token, maxsplit=1)
        reasoning_text = splits[0]
        normal_text = splits[1].strip()
        return StreamingParseResult(
            normal_text=normal_text, reasoning_text=reasoning_text)

스트리밍 파싱: parse_streaming_increment

토큰이 하나씩 들어올 때 증분 파싱을 수행한다. 태그가 불완전한 상태에서는 버퍼에 쌓아두고, 태그가 완성되면 처리한다.

def parse_streaming_increment(self, new_text: str) -> StreamingParseResult:
    self._buffer += new_text
    current_text = self._buffer
    think_start_text = self.think_start_token + self.think_start_self_label

    # 현재 텍스트가 think 토큰의 접두사이면 계속 버퍼링
    tokens_to_check = [think_start_text, self.think_end_token]

stream_reasoning=False이면 추론이 끝날 때까지 추론 내용을 축적했다가 한 번에 반환한다. True이면 추론 내용도 실시간으로 스트리밍한다.

Tool 호출 인터럽트

추론 중에 도구 호출 토큰(tool_start_token)이 나타나면 추론을 중단하고 도구 호출로 전환한다.

if (in_reasoning and self.tool_start_token is not None
    and self.tool_start_token in processed_text):
    tool_idx = processed_text.find(self.tool_start_token)
    reasoning_text = processed_text[:tool_idx].strip()
    normal_text = processed_text[tool_idx:]
    return StreamingParseResult(
        normal_text=normal_text, reasoning_text=reasoning_text)

StreamingParseResult

파싱 결과는 normal_textreasoning_text 두 필드로 구성된다.

class StreamingParseResult:
    def __init__(self, normal_text=None, reasoning_text=None):
        self.normal_text = normal_text or ""
        self.reasoning_text = reasoning_text or ""

Code Completion Parser: FIM 프롬프트

코드 완성은 FIM(Fill-in-the-Middle) 형식을 사용한다. CompletionTemplate은 모델별 FIM 토큰을 정의한다.

@dataclasses.dataclass
class CompletionTemplate:
    name: str
    fim_begin_token: str     # 예: <|fim_begin|>
    fim_middle_token: str    # 예: <|fim_middle|>
    fim_end_token: str       # 예: <|fim_end|>
    fim_position: FimPosition  # MIDDLE 또는 END

FimPosition은 middle 토큰의 위치를 결정한다. MIDDLE이면 begin + prefix + middle + suffix + end 순서, ENDbegin + prefix + end + suffix + middle 순서로 프롬프트를 구성한다.

class FimPosition(Enum):
    MIDDLE = auto()
    END = auto()

프롬프트 생성 함수는 suffix가 있을 때만 FIM 형식을 적용한다.

def generate_completion_prompt_from_request(request: CompletionRequest) -> str:
    if request.suffix == "":
        return request.prompt
    return generate_completion_prompt(
        request.prompt, request.suffix, completion_template_name)

템플릿 레지스트리

코드 완성 템플릿은 전역 레지스트리에 등록되며, 모델별로 자동 설정된다.

completion_templates: dict[str, CompletionTemplate] = {}

def register_completion_template(template, override=False):
    if not override:
        assert template.name not in completion_templates
    completion_templates[template.name] = template

def set_completion_template(template_name: str) -> None:
    global completion_template_name
    if completion_template_name is None:
        completion_template_name = template_name

파서 동작 비교

구분 일괄 파싱 스트리밍 파싱
입력 전체 텍스트 토큰 단위 증분
태그 처리 즉시 분리 버퍼링 후 처리
추론 출력 한 번에 반환 실시간 또는 축적
사용처 Non-streaming API Streaming API

관련 포스트

  • Server Args: 300+ 서버 인자 완전 가이드
  • Model Configuration 시스템: 모델 설정 관리

참고

  • 소스 코드: python/sglang/srt/parser/
  • 서버 인자 --reasoning-parser: 추론 파서 활성화
  • 서버 인자 --completion-template: 코드 완성 템플릿 지정

댓글

관련 포스트

SGLang 의 다른글