SGLang: LLM 서빙 프레임워크 설치 가이드 및 실습

import sglang as sgl import time # 서버 연결 sgl.set_default_backend(sgl.RuntimeEndpoint("http://localhost:30000")) # [비교군 1] FSM 최적화가 없는 일반 생성 (순수 모델 연산) @sgl.function def test_no_fsm(s, device): s += f"Extract the hardware specifications for {device} into a JSON format.\n" s += "Include keys: model_name, memory_size_gb, architecture.\n" # regex 제약 없음! GPU가 한 글자씩 모든 토큰의 확률을 계산해야 함 s += sgl.gen("json_spec", max_tokens=50) return s # [비교군 2] Compressed FSM 최적화 적용 (경로 압축 및 Jump-forward) @sgl.function def test_with_fsm(s, device): s += f"Extract the hardware specifications for {device} into a JSON format.\n" s += "Include keys: model_name, memory_size_gb, architecture.\n" # 정규표현식 제약 부여! SGLang 백엔드가 고정된 문자열은 GPU 연산 없이 한 번에 건너뜀 regex = r'\{\n "model_name": "[a-zA-Z0-9 -]+",\n "memory_size_gb": \d+,\n "architecture": "(Ampere|Ada|Hopper)"\n\}' s += sgl.gen("json_spec", max_tokens=50, regex=regex) return s print("▶ Compressed FSM 최적화 A/B 테스트 시작...\n") # 공정한 속도 비교를 위해 서로 다른 device 이름을 주어 각각 Cold Start 유도 print("==================================================") print("[Test A] Baseline: FSM 없는 일반 JSON 디코딩") start = time.perf_counter() res_no_fsm = test_no_fsm(device="NVIDIA RTX 3080") output_text = res_no_fsm['json_spec'] time_no_fsm = time.perf_counter() - start print(f"출력 결과:\n{output_text}\n") print(f"⏱️ 소요 시간: {time_no_fsm:.4f}초") print("==================================================\n") print("==================================================") print("[Test B] Optimized: Compressed FSM 기반 JSON 디코딩") start = time.perf_counter() res_fsm = test_with_fsm(device="NVIDIA RTX 4090") output_text = res_fsm['json_spec'] time_fsm = time.perf_counter() - start print(f"출력 결과:\n{output_text}\n") print(f"⏱️ 소요 시간: {time_fsm:.4f}초") print("==================================================\n") # 분모가 0이 되는 것을 방지 speed_up = time_no_fsm / time_fsm if time_fsm > 0 else 0 print(f"💡 시스템 최적화 분석: FSM 적용 시 연산량을 건너뛰어 약 {speed_up:.1f}배 더 빠릅니다!")

import sglang as sgl import time # 4세부 로컬 추론 엔진에 연결 sgl.set_default_backend(sgl.RuntimeEndpoint("http://localhost:30000")) @sgl.function def multi_dimensional_judge(s, topic, essay): # 1. 채팅 템플릿 적용 (System, User, Assistant 역할 부여) s += sgl.system("You are an expert essay evaluator.") s += sgl.user(f"Topic: {topic}\nEssay: {essay}\nIs the essay related to the topic? Answer strictly 'yes' or 'no'.") # 2. [최적화 1] 확률 마스킹을 통한 초고속 선택 (yes or no) s += sgl.assistant(sgl.gen("related", choices=["yes", "no"])) # 3. Python 제어 흐름 (Control Flow) 연동 if s["related"] == "no": s += sgl.user("Since it's not related, provide a brief rejection reason.") s += sgl.assistant(sgl.gen("rejection", max_tokens=30)) return s # 4. [최적화 2] Fork를 통한 병렬 평가 및 KV 캐시 재사용 (Sec. 3) dimensions = ["Clarity", "Logic", "Grammar"] s += sgl.user("Evaluate the essay based on the following dimensions.") forks = s.fork(len(dimensions)) for i, f in enumerate(forks): f += sgl.user(f"Evaluate based on: {dimensions[i]}. Provide a 1-sentence judgment. End with 'END'.") f += sgl.assistant("Judgment: " + sgl.gen("judgment", stop="END")) # 5. Join (병합) judgments = "\n".join([f"- {dimensions[i]}: {f['judgment']}" for i, f in enumerate(forks)]) # 6. [최적화 3] Compressed FSM을 통한 JSON 고속 제약 디코딩 (Sec. 4) s += sgl.user(f"Here are the judgments:\n{judgments}\nProvide a summary and a final letter grade (A, B, C, D, F) in JSON format.") # 강력한 정규식 제약 (이전 실습의 교훈 적용!) schema = r'\{\n "summary": "[a-zA-Z0-9 .,-]+",\n "grade": "(A|B|C|D|F)"\n\}' s += sgl.assistant(sgl.gen("output", regex=schema)) return s print("▶ SGLang 논문 Figure 2 (다차원 평가 에이전트) 통합 파이프라인 테스트...\n") test_topic = "The impact of Artificial Intelligence on future jobs." test_essay = "AI is rapidly evolving. It will automate repetitive tasks, but also create new tech jobs like prompt engineers. We must adapt through continuous learning." start = time.perf_counter() # 단 한 번의 호출로 저수준 백엔드 스케줄링까지 완벽하게 처리됩니다! state = multi_dimensional_judge(topic=test_topic, essay=test_essay) # 결과 출력 대기 (Blocking) final_json = state["output"] if state["related"] == "yes" else state["rejection"] total_time = time.perf_counter() - start print("================ [에이전트 판단 결과] ================") print(f"1. 주제 연관성 (yes/no): {state['related']}") if state["related"] == "yes": print(f"\n2. 최종 평가 JSON:\n{final_json}") print("======================================================") print(f"\n⏱️ 파이프라인 전체 소요 시간: {total_time:.4f}초")